o `j[@sdddlmZddlZddlZddlmZddlmZm Z e e Z GdddeZ GdddZdS))IntEnumN)FutureRecordMetadataFutureProduceResultc@seZdZdZdZdZdS) FinalStaterN)__name__ __module__ __qualname__ABORTEDFAILED SUCCEEDEDrr[/home/djax/ivt_ai_plugin/venv/lib/python3.10/site-packages/kafka/producer/producer_batch.pyr src@seZdZd(ddZeddZeddZedd Zed d Zed d Z eddZ d(ddZ ddZ ddZ ddZd)ddZddZd(ddZddZd(d d!Zed"d#Zd$d%Zd&d'ZdS)* ProducerBatchNcCsd|durtn|}d|_||_d|_d|_||_||_||_||_ t ||_ g|_ d|_ d|_dS)NrF)time monotonicmax_record_sizecreateddrainedattempts last_attempt last_appendrecordstopic_partitionrproduce_future_record_futures_retry _final_state)selftprnowrrr__init__s  zProducerBatch.__init__cC|jSN)rrrrr final_state"szProducerBatch.final_statecCs |jSr$)r next_offsetr%rrr record_count&s zProducerBatch.record_countcC|jr|jjSdSr$)r producer_idr%rrrr**zProducerBatch.producer_idcCr)r$)rproducer_epochr%rrrr,.r+zProducerBatch.producer_epochcCr)r$)r base_sequencer%rrrr-2r+zProducerBatch.base_sequencecCs|j}|duo |dkS)N)r-)rbase_seqrrr has_sequence6szProducerBatch.has_sequencec Cs|j||||}|durdS|durtn|}t|j|j|_||_t|j |j |j |j |dur5t |nd|dur>t |nd|rJtdd|Dnd}|j||S)Nr.css*|]\}}t|dt|VqdS)zutf-8N)lenencode).0h_keyh_valrrr Js(z+ProducerBatch.try_append..)rappendrrmaxrsizerrroffset timestampcrcr1sumr)r timestamp_mskeyvalueheadersr!metadatafuturerrr try_append;s" zProducerBatch.try_appendcsL|jdur td|jtj|_td|j|ddfdddS)z6Abort the batch and complete the future and callbacks.Nz3Batch has already been completed in final state: %sz#Aborting batch for partition %s: %sr.csSr$r)_ exceptionrrUsz%ProducerBatch.abort..) rErrorsIllegalStateErrorrr logdebugr_complete_future)rrGrrFrabortNs zProducerBatch.abortcCs|j||dS)asComplete the batch successfully. Arguments: base_offset (int): The base offset of the messages assigned by the server log_append_time (int): The log append time or -1 if CreateTime is being used Returns: True if the batch was completed as a result of this call, and False if it had been completed previously. ) base_offsetr>)done)rrOlog_append_timerrrcompleteWs zProducerBatch.completecCs0t|ts tdt|std|j||dS)a Complete the batch exceptionally. The provided top-level exception will be used for each record future contained in the batch. Arguments: top_level_exception (Exception): top-level partition error. record_exceptions_fn (callable int -> Exception): Record exception function mapping batch_index to the respective record exception. Raises: TypeError: if top_level_exception is not Exception, or record_exceptions_fn is not callable. Returns: True if the batch was completed as a result of this call, or False if it had been completed previously. z*top_level_exception must be type Exceptionz%record_exceptions_fn must be callable)top_level_exceptionrecord_exceptions_fn) isinstance Exception TypeErrorcallablerP)rrSrTrrrcomplete_exceptionallycs z$ProducerBatch.complete_exceptionallycCs|durtjntj}|jdur2||_|tjur td|j|n td|j||||||dS|jtjurW|tjurJtd||j||jdStd|j||j|dSt d|j|f) a Finalize the state of a batch. Final state, once set, is immutable. This function may be called once or twice on a batch. It may be called twice if 1. An inflight batch expires before a response from the broker is received. The batch's final state is set to FAILED. But it could succeed on the broker and second time around batch.done() may try to set SUCCEEDED final state. 2. If a transaction abortion happens or if the producer is closed forcefully, the final state is ABORTED but again it could succeed if broker responds with a success. Attempted transitions from [FAILED | ABORTED] --> SUCCEEDED are logged. Attempted transitions from one failure state to the same or a different failed state are ignored. Attempted transitions from SUCCEEDED to the same or a failed state throw an exception. Nz8Successfully produced messages to %s with base offset %szHFailed to produce messages to topic-partition %s with base offset %s: %sTzWProduceResponse returned %s for %s after batch with base offset %s had already been %s.zBIgnored state transition %s -> %s for %s batch with base offset %sz6A %s batch must not attempt another state change to %sF) rr r rrKrLrwarningrMrIrJ)rrOr>rSrTr&rrrrP{s*       zProducerBatch.donecCs(|jjr td|j|||fdS)NzBatch is already closed!)ris_donerIrJsuccess)rrOr>rTrrrrMs zProducerBatch._complete_futurecCs&|durtn|}|d||jkS)Ni)rrr)rdelivery_timeout_msr!rrrhas_reached_delivery_timeoutsz*ProducerBatch.has_reached_delivery_timeoutcCr#r$)rr%rrrin_retryszProducerBatch.in_retrycCs8|durtn|}d|_|jd7_||_||_dS)NTr)rrrrrr)rr!rrrretrys  zProducerBatch.retrycCs|jjSr$)rr[r%rrrr[szProducerBatch.is_donecCsd|j|jfS)Nz2ProducerBatch(topic_partition=%s, record_count=%d))rrr'r%rrr__str__s zProducerBatch.__str__cCs |j|jkSr$)r)rotherrrr__lt__s zProducerBatch.__lt__r$)NNNN)rr r r"propertyr&r(r*r,r-r0rDrNrRrYrPrMr^r_r`r[rarcrrrrrs6          (    r)enumrloggingr kafka.errorserrorsrIkafka.producer.futurerr getLoggerrrKrrrrrrs