o
    `j                     @   s\  d dl mZmZmZ d dlZd dlmZ d dlZd dlZd dl	Z	d dl
mZ d dlmZmZ d dlmZmZmZmZmZ d dlmZmZ eeZdZdZdZG dd	 d	ZG d
d deZ G dd deZ!G dd dZ"G dd dZ#G dd deZ$G dd de$Z%G dd de$Z&G dd de$Z'G dd de$Z(G dd de$Z)G dd de$Z*dS )     )ABCabstractmethodabstractpropertyN)IntEnum)FindCoordinatorRequestCoordinatorType)AddOffsetsToTxnRequestAddPartitionsToTxnRequestEndTxnRequestInitProducerIdRequestTxnOffsetCommitRequest)ConsumerGroupMetadataTopicPartitionc                   @   s<   e Zd ZdZdd Zedd Zdd Zdd	 Zd
d Z	dS )ProducerIdAndEpochproducer_idepochc                 C   s   || _ || _d S Nr   )selfr   r    r   `/home/djax/ivt_ai_plugin/venv/lib/python3.10/site-packages/kafka/producer/transaction_manager.py__init__   s   
zProducerIdAndEpoch.__init__c                 C   s
   t | jk S r   )NO_PRODUCER_IDr   r   r   r   r   is_valid       
zProducerIdAndEpoch.is_validc                 C   s   | j |j ko| j|jkS r   )r   r   producer_epoch)r   batchr   r   r   match$   s   zProducerIdAndEpoch.matchc                 C   s"   t |to| j|jko| j|jkS r   )
isinstancer   r   r   )r   otherr   r   r   __eq__'   s   "zProducerIdAndEpoch.__eq__c                 C   s   d | j| jS )Nz,ProducerIdAndEpoch(producer_id={}, epoch={}))formatr   r   r   r   r   r   __str__*   s   zProducerIdAndEpoch.__str__N)
__name__
__module____qualname__	__slots__r   propertyr   r   r"   r$   r   r   r   r   r      s    
r   c                   @   s<   e Zd ZdZdZdZdZdZdZdZ	dZ
d	Zed
d ZdS )TransactionStater                           c                 C   s   || j kr|| j| jfv S || jkr|| j | j| j| jfv S || jkr(|| jkS || jkr2|| jkS || jkr?|| j| jfv S || jkrP|| j| j| j| jfv S || jkrc|| j| j| j| j| jfv S || jkrjdS || jkrqdS d S NFT)	INITIALIZINGUNINITIALIZEDBUMPING_PRODUCER_EPOCHREADYCOMMITTING_TRANSACTIONABORTING_TRANSACTIONIN_TRANSACTIONABORTABLE_ERRORFATAL_ERROR)clssourcetargetr   r   r   is_transition_valid?   s4   













z$TransactionState.is_transition_validN)r%   r&   r'   r5   r4   r7   r:   r8   r9   r;   r<   r6   classmethodr@   r   r   r   r   r*   .   s    r*   c                   @   s   e Zd ZdZdZdZdZdS )Priorityr   r+   r,   r-   N)r%   r&   r'   FIND_COORDINATORINIT_PRODUCER_IDADD_PARTITIONS_OR_OFFSETSEND_TXNr   r   r   r   rB   a   s
    rB   c                   @   sD  e Zd ZdZdZdZddd	Zd
d Zdd Zdd Z	dd Z
dd Zdd Zdd Zdd Zdd Zdd ZdddZd d! Zd"d# Zd$d% Zed&d' Zd(d) Zd*d+ ZeejejejhZeejej ej!ej"hZ#d,Z$d-Z%d.Z&d/Z'd0Z(d1d2 Z)dd3d4Z*d5d6 Z+d7d8 Z,d9d: Z-d;d< Z.d=d> Z/d?d@ Z0dAdB Z1dCdD Z2dEdF Z3dGdH Z4dIdJ Z5dKdL Z6dMdN Z7dOdP Z8dQdR Z9dSdT Z:dUdV Z;dWdX Z<dYdZ Z=d[d\ Z>d]d^ Z?d_d` Z@dadb ZAdcdd ZBdedf ZCdgdh ZDdidj ZEdkdl ZFdmdn ZGdodp ZHdqdr ZIddsdtZJdudv ZKdwdx ZLdydz ZMd{d| ZNd}d~ ZOdd ZPdd ZQdd ZRdS )TransactionManagerzy
    A class which maintains state for transactions. Also keeps the state necessary to ensure idempotent production.
    r      Nr   d   r      c                 C   s   || _ || _tdd | _i | _|| _|| _d | _d | _	t
 | _t
 | _t
 | _t | _tj| _d | _ttt| _d| _g | _d| _| j| _|| _t | _ d S )Nc                   S   s   dS )Nr   r   r   r   r   r   <lambda>x   s    z-TransactionManager.__init__.<locals>.<lambda>Fr   )!_api_version	_metadatacollectionsdefaultdict_sequence_numbers_last_acked_offsettransactional_idtransaction_timeout_ms_transaction_coordinator_consumer_group_coordinatorset_new_partitions_in_transaction"_pending_partitions_in_transaction_partitions_in_transactiondict_pending_txn_offset_commitsr*   r5   _current_state_last_errorr   r   NO_PRODUCER_EPOCHproducer_id_and_epoch_transaction_started_pending_requests_pending_requests_sort_id"NO_INFLIGHT_REQUEST_CORRELATION_ID!_in_flight_request_correlation_idretry_backoff_ms	threading	Condition_lock)r   rS   rT   rf   api_versionmetadatar   r   r   r   t   s*   zTransactionManager.__init__c                 C   st   | j - |   | tj | ttt | j	
  t| | j}| | |jW  d    S 1 s3w   Y  d S r   )ri   _ensure_transactional_transition_tor*   r4   set_producer_id_and_epochr   r   r_   rQ   clearInitProducerIdHandlerrT   _enqueue_requestresultr   handlerr   r   r   initialize_transactions   s   

$z*TransactionManager.initialize_transactionsc                 C   s   | j > |  rtd| jtjkr	 W d   dS | tj | 	t
tt | j  t| d}| | W d   dS 1 sDw   Y  dS )a  Idempotent (non-transactional) producer: enqueue an InitProducerIdHandler.

        Drives UNINITIALIZED -> INITIALIZING; the handler completes the
        transition to READY on success. No-op outside UNINITIALIZED so
        repeated calls from the sender's run loop are safe.
        zyinit_producer_id is for idempotent (non-transactional) producers; use initialize_transactions for transactional producersNr   )ri   is_transactionalErrorsIllegalStateErrorr]   r*   r5   rm   r4   rn   r   r   r_   rQ   ro   rp   rq   rs   r   r   r   init_producer_id   s   

"z#TransactionManager.init_producer_idc                 C   sH   | j  |   |   | tj W d    d S 1 sw   Y  d S r   )ri   rl   _maybe_fail_with_errorrm   r*   r:   r   r   r   r   begin_transaction   s
   "z$TransactionManager.begin_transactionc                 C   sP   | j  |   |   | tj | dW  d    S 1 s!w   Y  d S NT)ri   rl   rz   rm   r*   r8   _begin_completing_transactionr   r   r   r   begin_commit   s   $zTransactionManager.begin_commitc                 C   sf   | j & |   | jtjkr|   | tj | j	  | 
dW  d    S 1 s,w   Y  d S NF)ri   rl   r]   r*   r;   rz   rm   r9   rX   ro   r}   r   r   r   r   begin_abort   s   
$zTransactionManager.begin_abortc                 C   s.   | j r
| |   t| |}| | |jS r   )rX   rq   &_add_partitions_to_transaction_handlerEndTxnHandlerrr   )r   	committedrt   r   r   r   r}      s
   

z0TransactionManager._begin_completing_transactionc                 C   s   t |trt|d}nt |tstdt|f |jdkr)|js)td|jf | j1 | 	  | 
  | jtjkr@tdtd||j t| ||}| | |jW  d   S 1 s`w   Y  dS )am  Send consumer-group offsets as part of the current transaction.

        Arguments:
            offsets ({TopicPartition: OffsetAndMetadata}): offsets to commit.
            group_metadata (ConsumerGroupMetadata or str): full group metadata
                from KafkaConsumer.group_metadata() (preferred - enables
                broker-side fencing per KIP-447), or a bare group_id string
                for backwards compatibility (broker treats it as v0-v2).

        Returns:
            FutureRecordMetadata-style Future that completes once the offsets
            are durably committed (or fails fatally).
        )group_idzjsend_offsets_to_transaction expects group_metadata to be a ConsumerGroupMetadata or a group_id str, got %rr   z]Invalid ConsumerGroupMetadata: generation_id=%s implies a joined group but member_id is emptyzWCannot send offsets to transaction because the producer is not in an active transactionz<Begin adding offsets %s for consumer group %s to transactionN)r    strr   	TypeErrortypegeneration_id	member_id
ValueErrorri   rl   rz   r]   r*   r:   rw   
KafkaErrorlogdebugr   AddOffsetsToTxnHandlerrq   rr   )r   offsetsgroup_metadatart   r   r   r   send_offsets_to_transaction   s0   



$z.TransactionManager.send_offsets_to_transactionc                 C   sr   | j , |   | |s| |r	 W d    d S td| | j| W d    d S 1 s2w   Y  d S )Nz,Begin adding new partition %s to transaction)ri   _fail_if_not_ready_for_sendis_partition_addedis_partition_pending_addr   r   rX   add)r   topic_partitionr   r   r   "maybe_add_partition_to_transaction   s   "z5TransactionManager.maybe_add_partition_to_transactionc                 C   s   | j 9 |  rtd| j|  r,|  std| jt	j
kr4td| jjf W d    d S W d    d S 1 s?w   Y  d S )NzmCannot perform send because at least one previous transactional or idempotent request has failed with errors.zdCannot perform a 'send' before completing a call to init_transactions when transactions are enabled.zCannot call send in state %s)ri   	has_errorrw   r   r^   rv   has_producer_idrx   r]   r*   r:   namer   r   r   r   r     s"   "z.TransactionManager._fail_if_not_ready_for_sendc                 C   sX   | j  |  r	 W d    dS |   p|| jv W  d    S 1 s%w   Y  d S r   )ri   has_fatal_errorrv   rZ   r   tpr   r   r   is_send_to_partition_allowed  s   $z/TransactionManager.is_send_to_partition_allowedc                 C   s   |d u r| j jS | j j|kS r   )r`   r   r   )r   r   r   r   r   r     s   z"TransactionManager.has_producer_idc                 C   
   | j d uS r   )rS   r   r   r   r   rv         
z#TransactionManager.is_transactionalc                 C   s>   | j  t| jpt| jW  d    S 1 sw   Y  d S r   )ri   boolrX   rY   r   r   r   r   has_partitions_to_add#     $z(TransactionManager.has_partitions_to_addc                 C   <   | j  | jtjtjfv W  d    S 1 sw   Y  d S r   )ri   r]   r*   r8   r9   r   r   r   r   is_completing'     $z TransactionManager.is_completingc                 C      | j S r   )r^   r   r   r   r   
last_error-     zTransactionManager.last_errorc                 C   r   r   )ri   r]   r*   r;   r<   r   r   r   r   r   1  r   zTransactionManager.has_errorc                 C   6   | j  | jtjkW  d    S 1 sw   Y  d S r   )ri   r]   r*   r6   r   r   r   r   is_bumping_epoch7     
$z#TransactionManager.is_bumping_epoch	RETRIABLE	ABORTABLEFATALNEEDS_EPOCH_BUMPNEEDS_PRODUCER_ID_RESETc                 C   s
   | j dkS )a  Return True if the broker supports InitProducerIdRequest v3+ (KIP-360).

        KIP-360 landed in Kafka 2.5. On older brokers we fall back to the
        pre-KIP-360 recovery: reset producer id for idempotent producers,
        fatal state for transactional producers.
        )r,   r/   )rM   r   r   r   r   _supports_epoch_bumpY  s   
z'TransactionManager._supports_epoch_bumpc                 C   s   t |tr|nt|}|| jv r| jS |tju r-|dur-|dkr-| |j}||kr-| jS || j	v rC| 
 r9| jS |  r@| jS | jS t|tjrL| jS |  rS| jS | jS )a  Categorize a batch-completion error into a recovery outcome.

        Used by the Sender to decide what to do with a failed batch. This
        method does not mutate any state - it is a pure classification
        helper. The caller is responsible for dispatching to the
        appropriate recovery path.

        Arguments:
            error (type or BaseException): The error class or instance.
            batch (ProducerBatch): The batch that failed.
            log_start_offset (int): log_start_offset from the broker's
                PartitionProduceResponse, or -1 if unknown / client-side
                failure. Used for KAFKA-5793 retention detection.

        Returns one of:
            ERROR_CLASS_RETRIABLE          - caller should retry the batch
            ERROR_CLASS_ABORTABLE          - transactional producer only;
                                              abort the transaction
            ERROR_CLASS_FATAL              - unrecoverable; transition to
                                              fatal error and fail the batch
            ERROR_CLASS_NEEDS_EPOCH_BUMP   - recoverable via KIP-360 epoch
                                              bump (only when broker supports
                                              InitProducerIdRequest v3+)
            ERROR_CLASS_NEEDS_PRODUCER_ID_RESET - non-transactional pre-KIP-360
                                                   fallback: reset the
                                                   producer id entirely

        Note: this classification is for transactional/idempotent producers
        only. Non-idempotent producers don't call this; the Sender uses
        simpler retry/fail logic for them.
        Nr   )r    r   _FATAL_ERRORSERROR_CLASS_FATALrw   UnknownProducerIdErrorlast_acked_offsetr   ERROR_CLASS_RETRIABLE_NEEDS_EPOCH_BUMP_ERRORSr   ERROR_CLASS_NEEDS_EPOCH_BUMPrv   #ERROR_CLASS_NEEDS_PRODUCER_ID_RESET
issubclassRetriableErrorERROR_CLASS_ABORTABLE)r   errorr   log_start_offset
error_type
last_ackedr   r   r   classify_batch_errorb  s$    

z'TransactionManager.classify_batch_errorc                 C   r   r   )ri   r]   r*   r9   r   r   r   r   is_aborting  r   zTransactionManager.is_abortingc                 C   sf   | j & | jtjkrtd| 	 W d    d S | jtj|d W d    d S 1 s,w   Y  d S )NzvSkipping transition to abortable error state since the transaction is already being  aborted. Underlying exception: %sr   )ri   r]   r*   r9   r   r   rm   r;   r   excr   r   r   transition_to_abortable_error  s   "z0TransactionManager.transition_to_abortable_errorc                 C   s<   | j  | jtj|d W d    d S 1 sw   Y  d S Nr   )ri   rm   r*   r<   r   r   r   r   transition_to_fatal_error  s   "z,TransactionManager.transition_to_fatal_errorc                 C   4   | j  || jv W  d    S 1 sw   Y  d S r   ri   rZ   r   	partitionr   r   r   r        $z%TransactionManager.is_partition_addedc                 C   s   || j v p	|| jv S r   )rX   rY   r   r   r   r   r        z+TransactionManager.is_partition_pending_addc                 C   s   | j j|ko| j j|kS r   )r`   r   r   )r   r   r   r   r   r   has_producer_id_and_epoch  s   
z,TransactionManager.has_producer_id_and_epochc                 C   s.   t |ts	tdtd|j|j || _d S )Nz ProducerAndIdEpoch type requiredz"ProducerId set to %s with epoch %s)r    r   r   r   infor   r   r`   )r   r`   r   r   r   rn     s   

z,TransactionManager.set_producer_id_and_epochc                 C   sb   | j $ |  rtd| ttt | j	  | j
	  W d   dS 1 s*w   Y  dS )a  
        This method is used when the producer needs to reset its internal state because of an irrecoverable exception
        from the broker.

        We need to reset the producer id and associated state when we have sent a batch to the broker, but we either get
        a non-retriable exception or we run out of retries, or the batch expired in the producer queue after it was already
        sent to the broker.

        In all of these cases, we don't know whether batch was actually committed on the broker, and hence whether the
        sequence number was actually updated. If we don't reset the producer state, we risk the chance that all future
        messages will return an OutOfOrderSequenceNumberError.

        Note that we can't reset the producer state for the transactional producer as this would mean bumping the epoch
        for the same producer id. This might involve aborting the ongoing transaction during the initProducerIdRequest,
        and the user would not have any way of knowing this happened. So for the transactional producer,
        it's best to return the produce error to the user and let them abort the transaction and close the producer explicitly.
        zCannot reset producer state for a transactional producer. You must either abort the ongoing transaction or reinitialize the transactional producer insteadN)ri   rv   rw   rx   rn   r   r   r_   rQ   ro   rR   r   r   r   r   reset_producer_id  s   
"z$TransactionManager.reset_producer_idc                 C   s   | j j | jtjkr	 W d   dS | jtjkr"	 W d   dS |  s/td| jf t	
d| j | tj | j  | j  d| _| j  | j  | j  t| | jdd}| | W d   dS 1 spw   Y  dS )a  KIP-360: recover from a transient producer-state error by bumping
        the epoch.

        Transitions to BUMPING_PRODUCER_EPOCH and enqueues an
        InitProducerIdRequest v3+ carrying the current producer_id/epoch.
        When the broker responds with the bumped epoch, _complete_epoch_bump
        transitions back to READY and the sender resumes producing under
        the new epoch. Records in the accumulator that haven't been drained
        yet will be stamped with the new epoch on the next drain.

        TODO (KAFKA-5793 full): in-flight batches at the moment of the bump
        are lost--their futures fail. Adding in-place rewrite of the
        closed batch buffer (producer_id/epoch/base_sequence fields + CRC
        recompute) would let us retry them under the new epoch without
        losing records.

        Requires broker >= 2.5 (InitProducerIdRequest v3+). On older
        brokers, Sender falls back to reset_producer_id / fatal instead
        via classify_batch_error.

        Idempotent: if we're already in BUMPING_PRODUCER_EPOCH, this is a
        no-op. This matters because with max_in_flight > 1, multiple
        in-flight batches may all fail with the same epoch-bump-triggering
        error in quick succession; only the first should drive the bump.
        NzvCannot bump producer epoch: broker version %s does not support KIP-360 (InitProducerIdRequest v3+ requires Kafka 2.5+)z5Bumping producer epoch for %s after recoverable errorFTis_epoch_bump)ri   r]   r*   r6   r<   r   rw   rx   rM   r   warningr`   rm   rQ   ro   rR   ra   rZ   rX   rY   rp   rT   rq   rs   r   r   r   bump_producer_id_and_epoch  s4   




"z-TransactionManager.bump_producer_id_and_epochc                 C   s   |  tj d| _dS )zCalled from InitProducerIdHandler on successful bump response.

        Transitions BUMPING_PRODUCER_EPOCH -> READY so the sender resumes
        producing under the new epoch.
        N)rm   r*   r7   r^   r   r   r   r   _complete_epoch_bump   s   
z'TransactionManager._complete_epoch_bumpc                 C   s2   |  ttt t| |dd}||_| | dS )a  Called from InitProducerIdHandler when the broker rejects the bump
        with INVALID_PRODUCER_EPOCH (our producer_id/epoch are stale).

        Falls back to requesting a fresh producer_id by enqueuing a new
        InitProducerIdRequest without the producer_id/epoch fields. The
        original TransactionalRequestResult is re-used so the caller waits
        on the overall bump-then-init sequence.
        Fr   N)rn   r   r   r_   rp   _resultrq   )r   rT   rr   rt   r   r   r   '_restart_epoch_bump_without_producer_id*  s   
z:TransactionManager._restart_epoch_bump_without_producer_idc                 C   s4   | j  | j| W  d    S 1 sw   Y  d S r   ri   rQ   r   r   r   r   sequence_number;  r   z"TransactionManager.sequence_numberc                 C   s   | j ; || jvrtd| j| }|d| kr%|d|  d | j|< n| j|  |7  < W d    d S W d    d S 1 sAw   Y  d S )NzNAttempt to increment sequence number for a partition with no current sequence.ir+   )ri   rQ   rw   rx   )r   r   	incrementbaser   r   r   increment_sequence_number?  s   


"z,TransactionManager.increment_sequence_numberc                 C   s6   | j  || j|< W d    d S 1 sw   Y  d S r   r   )r   r   sequencer   r   r   set_sequence_numberJ  s   "z&TransactionManager.set_sequence_numberc                 C   sH   | j  | j|d  | j|d  W d    d S 1 sw   Y  d S r   )ri   rQ   poprR   r   r   r   r   reset_sequence_for_partitionN  s   "z/TransactionManager.reset_sequence_for_partitionc                 C   sp   |dk rdS || d }| j  || j|dkr&|| j|< W d   dS W d   dS 1 s1w   Y  dS )aK  Record the offset of the last successfully-produced record for tp.

        Called from the sender on each successful batch completion. The
        last acked offset is used to detect whether a subsequent
        UnknownProducerIdError reflects retention (safe to retry) vs. real
        data loss (fatal). See KAFKA-5793.
        r   Nr+   r   ri   rR   get)r   r   base_offsetrecord_countlast_offsetr   r   r   update_last_acked_offsetS  s   "z+TransactionManager.update_last_acked_offsetc                 C   s8   | j  | j|dW  d    S 1 sw   Y  d S )Nr   r   r   r   r   r   r   b  s   $z$TransactionManager.last_acked_offsetc              	   C   s8  | j  | jr| |   | js	 W d    d S | jd \}}}t|tr2|r2	 W d    d S t| j | 	|rMt
d|j 	 W d    d S t|tr| js|j  | jtjkrit
d |   zt| j\}}}W n ty   d }Y nw |rt
d|j |W  d    S 1 sw   Y  d S )Nr   zENot sending transactional request %s because we are in an error statezcNot sending EndTxn for completed transaction since no partitions or offsets were successfully addedzRequest %s dequeued for sending)ri   rX   rq   r   rb   r    r   heapqheappop#_maybe_terminate_request_with_errorr   r   requestra   rr   doner]   r*   r<   _complete_transaction
IndexError)r   has_incomplete_batches_next_request_handlerr   r   r   r   f  s>   


$z'TransactionManager.next_request_handlerc                 C   s>   | j  |  | | W d    d S 1 sw   Y  d S r   )ri   	set_retryrq   r   r   r   r   r   retry  s   "zTransactionManager.retryc                 C   sH   | j  | jD ]
\}}}|| qW d    d S 1 sw   Y  d S r   )ri   rb   fatal_error)r   r   r   r   r   r   r   authentication_failed  s
   "z(TransactionManager.authentication_failedc                 C   s0   |t jkr| jS |t jkr| jS td|f )Nz(Received an invalid coordinator type: %s)r   GROUPrV   TRANSACTIONrU   rw   rx   )r   
coord_typer   r   r   coordinator  s
   

zTransactionManager.coordinatorc                 C   s   |  |j|j d S r   )_lookup_coordinatorcoordinator_typecoordinator_keyr   r   r   r   lookup_coordinator_for_request  r   z1TransactionManager.lookup_coordinator_for_requestc                 C      |  j d7  _ | j S Nr+   )re   r   r   r   r   %next_in_flight_request_correlation_id     z8TransactionManager.next_in_flight_request_correlation_idc                 C   s   | j | _d S r   )rd   re   r   r   r   r   4clear_in_flight_transactional_request_correlation_id     zGTransactionManager.clear_in_flight_transactional_request_correlation_idc                 C   s   | j | jkS r   )re   rd   r   r   r   r   #has_in_flight_transactional_request  r  z6TransactionManager.has_in_flight_transactional_requestc                 C      | j tjkS r   )r]   r*   r<   r   r   r   r   r     r  z"TransactionManager.has_fatal_errorc                 C   r  r   )r]   r*   r;   r   r   r   r   has_abortable_error  r  z&TransactionManager.has_abortable_errorc                 C   r   r   r   r   r   r   r   $_test_transaction_contains_partition  r   z7TransactionManager._test_transaction_contains_partitionc                 C   s
   t | jS r   )r   r\   r   r   r   r    _test_has_pending_offset_commits  r   z3TransactionManager._test_has_pending_offset_commitsc                 C   sF   | j  | jtjkp|  p|  W  d    S 1 sw   Y  d S r   )ri   r]   r*   r:   r   r  r   r   r   r   _test_has_ongoing_transaction  s   $z0TransactionManager._test_has_ongoing_transactionc                 C   s>   | j  |  o| jtjkW  d    S 1 sw   Y  d S r   )ri   rv   r]   r*   r7   r   r   r   r   _test_is_ready  r   z!TransactionManager._test_is_readyc                 C   s   | j X | j| j|std| j| jj|jf |tjtj	fv r3|d u r/t
d|jf || _nd | _| jd urHtd| jj|j| j ntd| j| || _W d    d S 1 s^w   Y  d S )NzJTransactionalId %s: Invalid transition attempted from state %s to state %sz.Cannot transition to %s with an None exceptionz/Transition from state %s to error state %s (%s)zTransition from state %s to %s)ri   r]   r@   rw   r   rS   r   r*   r<   r;   IllegalArgumentErrorr^   r   r   )r   r?   r   r   r   r   rm     s   
"z!TransactionManager._transition_toc                 C   s   |   s	tdd S )Nz=Transactional method invoked on a non-transactional producer.)rv   rw   rx   r   r   r   r   rl     s   
z(TransactionManager._ensure_transactionalc                 C   s   |   rtd| jf d S )NzHCannot execute transactional method because we are in an error state: %s)r   rw   r   r^   r   r   r   r   rz     s   z)TransactionManager._maybe_fail_with_errorc                 C   s2   |   r|  rt|trdS || j dS dS r3   )r   r  r    FindCoordinatorHandlerfailr^   r   request_handlerr   r   r   r     s   z6TransactionManager._maybe_terminate_request_with_errorc                 C   r   r  )rc   r   r   r   r   _next_pending_requests_sort_id  r  z1TransactionManager._next_pending_requests_sort_idc                 C   s,   t d|j t| j|j|  |f d S )Nz"Enqueuing transactional request %s)r   r   r   r   heappushrb   priorityr  r  r   r   r   rq     s   z#TransactionManager._enqueue_requestc                 C   sp   | j " |tjkrd | _n|tjkrd | _ntd|f W d    n1 s(w   Y  | t	| || d S )NzInvalid coordinator type: %s)
ri   r   r   rV   r   rU   rw   rx   rq   r  )r   r   	coord_keyr   r   r   r     s   

z&TransactionManager._lookup_coordinatorc                 C   s\   | j ! | tj d| _| j  | j  | j  W d    d S 1 s'w   Y  d S r   )	ri   rm   r*   r7   ra   rX   ro   rY   rZ   r   r   r   r   r     s   

"z(TransactionManager._complete_transactionc                 C   sN   | j  | j| j | j  t| | jW  d    S 1 s w   Y  d S r   )ri   rY   updaterX   ro   AddPartitionsToTxnHandlerr   r   r   r   r     s
   

$z9TransactionManager._add_partitions_to_transaction_handler)Nr   rI   rJ   Nr   )r   )Sr%   r&   r'   __doc__rd   ADD_PARTITIONS_RETRY_BACKOFF_MSr   ru   ry   r{   r~   r   r}   r   r   r   r   r   rv   r   r   r)   r   r   r   	frozensetrw   OutOfOrderSequenceNumberErrorr   InvalidProducerEpochErrorr   ClusterAuthorizationFailedError'TransactionalIdAuthorizationFailedErrorProducerFencedErrorInvalidTxnStateErrorr   r   r   r   r   r   r   r   r   r   r   r   r   r   rn   r   r   r   r   r   r   r   r   r   r   r   r   r   r   r   r  r  r  r   r  r	  r
  r  r  rm   rl   rz   r   r  rq   r   r   r   r   r   r   r   rG   k   s    
$
%


	
	F8
#
	
rG   c                   @   sX   e Zd Zdd ZdddZdddZedd	 Zed
d Zedd Z	edd Z
dS )TransactionalRequestResultc                 C   s   t  | _d | _d S r   )rg   Event_latch_errorr   r   r   r   r     s   

z#TransactionalRequestResult.__init__Nc                 C   s   || _ | j  d S r   )r$  r#  rW   )r   r   r   r   r   r     s   zTransactionalRequestResult.donec                 C   s0   |d ur|d nd }| j |}| jr| j|S )Ni  )r#  waitr$  )r   
timeout_mstimeoutsuccessr   r   r   r%    s
   zTransactionalRequestResult.waitc                 C   s
   | j  S r   )r#  is_setr   r   r   r   is_done  r   z"TransactionalRequestResult.is_donec                 C   s   | j  o	| jd u S r   r#  r)  r$  r   r   r   r   	succeeded"     z$TransactionalRequestResult.succeededc                 C   s   | j  o	| jd uS r   r+  r   r   r   r   failed&  r-  z!TransactionalRequestResult.failedc                 C   r   r   )r$  r   r   r   r   	exception*  r   z$TransactionalRequestResult.exceptionr   )r%   r&   r'   r   r   r%  r)   r*  r,  r.  r/  r   r   r   r   r!    s    




r!  c                   @   s   e Zd Zd$ddZedd Zedd Zedd	 Zd
d Zdd Z	dd Z
dd Zdd Zdd Zedd Zedd Zedd Zdd Zedd Zed d! Zed"d# ZdS )%TxnRequestHandlerNc                 C   s*   || _ |j| _d | _|pt | _d| _d S r   )transaction_managerrf   r   r!  r   	_is_retry)r   r1  rr   r   r   r   r   0  s
   
zTxnRequestHandler.__init__c                 C      | j jS r   r1  rS   r   r   r   r   rS   7     z"TxnRequestHandler.transactional_idc                 C   
   | j jjS r   )r1  r`   r   r   r   r   r   r   ;  r   zTxnRequestHandler.producer_idc                 C   r6  r   )r1  r`   r   r   r   r   r   r   ?  r   z TxnRequestHandler.producer_epochc                 C   sB   t d| jr
| jjnd d|  | j| | jj|d d S )NzFatal Error handling request nonez: r   )r   r   r   r   r1  r   r   r   r   r   r   r   r   C  s   $zTxnRequestHandler.fatal_errorc                 C   s   | j | | jj|d d S r   )r1  r   r   r   r   r   r   r   abortable_errorH  s   z!TxnRequestHandler.abortable_errorc                 C   s   | j j|d d S r   )r   r   r   r   r   r   r  L  s   zTxnRequestHandler.failc                 C   s@   | j j d| _| j |  W d    d S 1 sw   Y  d S r|   )r1  ri   r2  rq   r   r   r   r   	reenqueueO  s   
"zTxnRequestHandler.reenqueuec                 C   s   || j jkr| td d S | j   t|tjr2t	d | 
 r,| j | j| j |   d S t|tjr?| | d S t|ttd fslt	d|| j | j j | | W d    d S 1 sew   Y  d S | td|  d S )Nz7Detected more than one in-flight transactional request.z#Disconnected from node. Will retry.z1Received transactional response %s for request %sz?Could not execute transactional request for unknown reasons: %s)r1  re   r   RuntimeErrorr  r    rw   KafkaConnectionErrorr   r   needs_coordinatorr   r   r   r9  UnsupportedVersionError	Exceptionr   r   ri   handle_responser   )r   correlation_idresponse_or_excr   r   r   on_completeT  s    


"zTxnRequestHandler.on_completec                 C   r   r   )r   r   r   r   r   r<  g  r   z#TxnRequestHandler.needs_coordinatorc                 C   r   r   )r   r   r   r   r   rr   j  r   zTxnRequestHandler.resultc                 C      t jS r   )r   r   r   r   r   r   r   n  r   z"TxnRequestHandler.coordinator_typec                 C   r3  r   r4  r   r   r   r   r   r  r5  z!TxnRequestHandler.coordinator_keyc                 C   s
   d| _ d S r|   r2  r   r   r   r   r   v  r   zTxnRequestHandler.set_retryc                 C   r   r   rD  r   r   r   r   is_retryy  r   zTxnRequestHandler.is_retryc                 C      d S r   r   )r   responser   r   r   r?  }     z!TxnRequestHandler.handle_responsec                 C   rF  r   r   r   r   r   r   r    rH  zTxnRequestHandler.priorityr   )r%   r&   r'   r   r)   rS   r   r   r   r8  r  r9  rB  r<  rr   r   r   r   rE  r   r?  r   r  r   r   r   r   r0  /  s6    








r0  c                       s>   e Zd Zd
 fdd	Zedd Zedd Zdd	 Z  ZS )rp   Fc                    sX   t  | || _d}d}|rd}|jj}|jj}nt}t}t| j	|||||d| _
d S )Nr.   r   r-   )rS   rT   r   r   max_versionmin_version)superr   _is_epoch_bumpr`   r   r   r   r_   r   rS   r   )r   r1  rT   r   rI  rJ  r   r   	__class__r   r   r     s$   
zInitProducerIdHandler.__init__c                 C   rC  r   )rB   rD   r   r   r   r   r    r   zInitProducerIdHandler.priorityc                 C   s   | j jd u rd S tjS r   )r1  rS   r   r   r   r   r   r   r     s   z&InitProducerIdHandler.coordinator_typec                 C   s  t |j}|t ju r-| jt|j|j | j	r| j
  n| jtj | j  d S t|t jrJ|t jt jfv rD| jtj| j |   d S |t ju rc| j	rctd | j| jj| j d S |t jt jfv rt|  t   d S |t j!u r|  |  d S |  t "d|   d S )Nz]InitProducerId bump rejected with INVALID_PRODUCER_EPOCH; falling back to a fresh producer idz.Unexpected error in InitProducerIdResponse: %s)#rw   for_code
error_codeNoErrorr1  rn   r   r   r   rL  r   rm   r*   r7   r   r   r   r   NotCoordinatorErrorCoordinatorNotAvailableErrorr   r   r   rS   r9  r  r   r   r   r   rT   r  r   r  r   r   rG  r   r   r   r   r?    s*   



z%InitProducerIdHandler.handle_response)F)	r%   r&   r'   r   r)   r  r   r?  __classcell__r   r   rM  r   rp     s    

rp   c                       s8   e Zd Z fddZedd Zdd Zdd Z  ZS )	r  c                    sh   t  | tt}|D ]}||j |j qtj	 t| j
| j| j fdd| D dd| _d S )Nc                       g | ]
\}} ||d qS )r   
partitionsr   .0topicrX  Topicr   r   
<listcomp>      z6AddPartitionsToTxnHandler.__init__.<locals>.<listcomp>r-   )v3_and_below_transactional_idv3_and_below_producer_idv3_and_below_producer_epochv3_and_below_topicsrI  )rK  r   rO   rP   listr[  appendr   r	   AddPartitionsToTxnTopicrS   r   r   itemsr   )r   r1  topic_partitions
topic_datar   rM  r\  r   r     s   

z"AddPartitionsToTxnHandler.__init__c                 C   rC  r   rB   rE   r   r   r   r   r    r   z"AddPartitionsToTxnHandler.priorityc                 C   s  d}t  }| jj| _dd |jD }| D ]\}}|tju r qt|tjrH|tj	tj
fv r8| jtj| j n	|tju rA|   |    d S |tjtjfv rZ| t   d S |tju rh| |   d S |tjtjfv r|| t|   d S |tju r||j q|tju rtd| d}qtd||  d}qt |}| j j |8  _ |r| !t| d S |r| !td|  d S td| | jj"#| d| j_$| j%&  d S )	NFc                 S   0   i | ]\}}|D ]\}}t ||t|qqS r   r   rw   rO  rZ  r[  partition_datar   rP  r   r   r   
<dictcomp>      z=AddPartitionsToTxnHandler.handle_response.<locals>.<dictcomp>zdDid not attempt to add partition %s to transaction because other partitions in the batch had errors.Tz5Could not add partition %s due to unexpected error %sz9Could not add partitions to transaction due to errors: %sz/Successfully added partitions %s to transaction)'rW   r1  rf   results_by_topic_v3_and_belowrg  rw   rQ  r   r   rS  rR  r   r   r   rS   ConcurrentTransactionsErrormaybe_override_retry_backoff_msr9  r  r  r   r  InvalidProducerIdMappingErrorr   r   TopicAuthorizationFailedErrorr   r[  OperationNotAttemptedErrorr   r   r   rY   r8  rZ   r  ra   r   r   )r   rG  has_partition_errorsunauthorized_topicsresultsr   r   rX  r   r   r   r?    sX   





z)AddPartitionsToTxnHandler.handle_responsec                 C   s"   | j jst| j j| j| _d S d S r   )r1  rZ   minr  rf   r   r   r   r   rs    s   z9AddPartitionsToTxnHandler.maybe_override_retry_backoff_ms)	r%   r&   r'   r   r)   r  r?  rs  rU  r   r   rM  r   r    s    
;r  c                       sH   e Zd Z fddZedd Zedd Zedd Zd	d
 Z  Z	S )r  c                    s<   t  | t|| _|| _t| j| jj| jgd| _d S )N)keykey_typecoordinator_keys)	rK  r   r   
build_from_coord_type
_coord_keyr   valuer   )r   r1  r   r  rM  r   r   r   +  s   zFindCoordinatorHandler.__init__c                 C   rC  r   )rB   rC   r   r   r   r   r  9  r   zFindCoordinatorHandler.priorityc                 C   rF  r   r   r   r   r   r   r   =  rH  z'FindCoordinatorHandler.coordinator_typec                 C   rF  r   r   r   r   r   r   r   A  rH  z&FindCoordinatorHandler.coordinator_keyc                 C   s   |j r|j d n|}t|j}|tju r<| jj|| j| j	}| jt
jkr+|| j_n
| jt
jkr5|| j_| j  d S t|tjrH|   d S |tju rU| |  d S |tju rd| || j	 d S | td| j| j	| f  d S )Nr   zQCould not find a coordinator with type %s with key %s due to unexpected error: %s)coordinatorsrw   rO  rP  rQ  r1  rN   add_coordinatorr  r  r   r   rV   r   rU   r   r   r   r   r9  r  r   GroupAuthorizationFailedErrorr8  r   )r   rG  rr   r   coordinator_idr   r   r   r?  E  s,   





z&FindCoordinatorHandler.handle_response)
r%   r&   r'   r   r)   r  r   r   r?  rU  r   r   rM  r   r  *  s    


r  c                       0   e Zd Z fddZedd Zdd Z  ZS )r   c                    s*   t  | t| j| j| j|dd| _d S )Nr-   )rS   r   r   r   rI  )rK  r   r
   rS   r   r   r   )r   r1  r   rM  r   r   r   `  s   zEndTxnHandler.__init__c                 C   rC  r   )rB   rF   r   r   r   r   r  i  r   zEndTxnHandler.priorityc                 C   s   t |j}|t ju r| j  | j  d S t|t j	r4|t j
t jfv r.| jtj| j |   d S |t jt jfv rE| t   d S |t ju rR| |  d S |t ju r_| |  d S | t d|   d S )Nz%Unhandled error in EndTxnResponse: %s)rw   rO  rP  rQ  r1  r   r   r   r   r   rS  rR  r   r   r   rS   r9  r  r  r   r  r   r   rT  r   r   r   r?  m  s   



zEndTxnHandler.handle_responser%   r&   r'   r   r)   r  r?  rU  r   r   rM  r   r   _  s
    	
r   c                       r  )r   c                    s@   t  | || _|j| _|| _t| j| j| j	| jdd| _
d S )Nr-   )rS   r   r   r   rI  )rK  r   r   r   consumer_group_idr   r   rS   r   r   r   )r   r1  r   r   rM  r   r   r     s   zAddOffsetsToTxnHandler.__init__c                 C   rC  r   rj  r   r   r   r   r    r   zAddOffsetsToTxnHandler.priorityc                 C   sx  t |j}|t ju r:td| j | j D ]
\}}|| j	j
|< qt| j	| j| j	j
| j}| j	| d| j	_d S t|t jrW|t jt jfv rQ| j	tj| j |   d S |t jt jfv re|   d S |t jt jfv rv| t   d S |t jt jfv r| j	  r| !|  d S | |  d S |t j"u r| |  d S |t j#u r| !|| j d S | t $d|   d S )NzASuccessfully added partition for consumer group %s to transactionTz/Unexpected error in AddOffsetsToTxnResponse: %s)%rw   rO  rP  rQ  r   r   r  r   rg  r1  r\   TxnOffsetCommitHandlerr   r   rq   ra   r   r   rS  rR  r   r   r   rS   r9  CoordinatorLoadInProgressErrorrr  r  r  r   r   rt  r   r8  r  r  r   )r   rG  r   r   offsetrt   r   r   r   r?    s6   





z&AddOffsetsToTxnHandler.handle_responser  r   r   rM  r   r     s
    
r   c                       sP   e Zd Z fddZdd Zedd Zedd Zed	d
 Zdd Z	  Z
S )r  c                    s2   t  j||d || _|j| _|| _|  | _d S )N)rr   )rK  r   r   r   r  r   _build_requestr   )r   r1  r   r   rr   rM  r   r   r     s
   zTxnOffsetCommitHandler.__init__c                    s   t j  j}tt}| j D ]\}}||j 	||j
|j|j|jd qt | j| j| j| j| jj| jj| jj fdd| D dd	S )N)partition_indexcommitted_offsetcommitted_leader_epochcommitted_metadatac                    rV  rW  r   rY  r\  r   r   r^    r_  z9TxnOffsetCommitHandler._build_request.<locals>.<listcomp>r-   )	rS   r   r   r   r   r   group_instance_idtopicsrI  )r   TxnOffsetCommitRequestTopicTxnOffsetCommitRequestPartitionrO   rP   rd  r   rg  r[  re  r   r  leader_epochrk   rS   r  r   r   r   r   r   r  )r   	Partitionri  r   r  r   r\  r   r    s.   


z%TxnOffsetCommitHandler._build_requestc                 C   rC  r   rj  r   r   r   r   r    r   zTxnOffsetCommitHandler.priorityc                 C   rC  r   )r   r   r   r   r   r   r     r   z'TxnOffsetCommitHandler.coordinator_typec                 C   r   r   )r  r   r   r   r   r     r   z&TxnOffsetCommitHandler.coordinator_keyc                 C   s  d}d}dd |j D }| D ]\}}|tju r'td|| j | jj|= qt	|tj
r<d}|tjtjtjfv r;d}q|tju rL| || j  d S |tjtjfv r^| t   d S |tju rl| |   d S |tjtjfv r| td|jf   d S |tjtjfv r| |   d S | td|    d S |r| jtj| j |s| j  d S | jjr| jj| _ | ! | _"| #  d S d S )NFc                 S   rk  r   rl  rm  r   r   r   ro    rp  z:TxnOffsetCommitHandler.handle_response.<locals>.<dictcomp>zHSuccessfully added offsets for %s from consumer group %s to transaction.TzLTransaction offset commit failed due to consumer group metadata mismatch: %sz/Unexpected error in TxnOffsetCommitResponse: %s)$r  rg  rw   rQ  r   r   r  r1  r\   r   r   rS  rR  RequestTimedOutErrorr  r8  r  r  r   FencedInstanceIdErrorIllegalGenerationErrorUnknownMemberIdErrorCommitFailedErrorr%   r   UnsupportedForMessageFormatErrorr   r   r   r   rr   r   r   r  r   r9  )r   rG  lookup_coordinatorretriable_failureerrorsr   r   r   r   r   r?    sd   




z&TxnOffsetCommitHandler.handle_response)r%   r&   r'   r   r  r)   r  r   r   r?  rU  r   r   rM  r   r    s    


r  )+abcr   r   r   rO   enumr   r   loggingrg   kafka.errorsr  rw   kafka.protocol.metadatar   r   kafka.protocol.producerr   r	   r
   r   r   kafka.structsr   r   	getLoggerr%   r   r   r_   NO_SEQUENCEr   r*   rB   rG   r!  r0  rp   r  r  r   r   r  r   r   r   r   <module>   s>    
3
     (!WI[5$9