o `j@sddlZddlZddlZddlZddlZddlZddlmZddl m Z ddl m Z m Z mZddlmZddlmZmZddlmZddlmZdd lmZeeZejjZGd d d ej Z!Gd d d Z"dS)N)errors)AnonMeasurable)AvgMaxRate)TransactionManager)ProduceRequestProduceResponse)TopicPartition)ensure_valid_topic_name) __version__c seZdZdZddedddddddd d ed Zfd d ZddZd@ddZ ddZ ddZ d@ddZ ddZ ddZddZddZd d!Zd"d#Zd$d%Zd&d'Zd(d)Zd*d+Zd,d-Zd.d/Zd0d1Zd2d3Zd4d5Zd6d7Zd8d9Zd:d;Zdd?Z Z!S)ASenderz The background thread that handles the sending of produce requests to the Kafka cluster. This thread makes metadata requests to renew its view of the cluster and then sends produce requests to the appropriate nodes. iinfi0udNFi`z kafka-python-) max_request_sizeacksretriesrequest_timeout_msretry_backoff_msmetricsguarantee_message_ordertransaction_managertransactional_idtransaction_timeout_ms client_idc stt|j|_|jD]}||vr|||j|<q|jdd|_||_||_|j |_ d|_ d|_ t |_|jdrLt|jd|j|j |_nd|_|jd|_tt|_dS)Nrz-network-threadTFrr)super__init__copyDEFAULT_CONFIGconfigpopname_client _accumulatorcluster _metadata_running _force_closeset_topics_to_add SenderMetrics_sensors_transaction_manager collections defaultdictlist_in_flight_batches)selfclientmetadata accumulatorconfigskey __class__S/home/djax/ivt_ai_plugin/venv/lib/python3.10/site-packages/kafka/producer/sender.pyr/s$    zSender.__init__cCsrz|j|j}Wn tyYdSwz ||j|f}Wn ty'YdSw|d||<|t|dS)N) r1topic_partitionKeyErrorindexcreated ValueErrorr!heapqheapify)r2batchqueueidxr:r:r;#_maybe_remove_from_inflight_batchesEs   z*Sender._maybe_remove_from_inflight_batchescCsg}g}|jD]>\}}|rB|d\}}||jjr9t||jdur,||nt d|j |j |jf|j |n|s||q |D]}|j|=qJ|S)zz.Sender._send_producer_data..TzCExpiring %d record(s) for %s: %s ms has passed since batch creationrz%%s: Nodes with data ready to send: %sz#%s: Created %d produce requests: %sz%s: Sending Produce Request: %r)2time monotonicr$readyr&rWrXrYrequest_updatefloatr0r#is_readyconnection_delay maybe_connectremovemindrainr valuesr@r1r=rBheappushmutedaddreset_next_batch_expiry_timerRextendrVlenr-any record_countint_complete_batch_with_exceptionrNKafkaTimeoutErrorr,update_produce_request_metricsreset_producer_id_create_produce_requestsnext_expiry_time_msrHsend add_callback_handle_produce_response add_errback_failed_produce)r2rQresult ready_nodesnext_ready_check_delayunknown_leaders_existnot_ready_timeout_msnode node_delay_msbatches_by_node batch_listrDitemrErRneeds_transaction_state_reset expired_batch error_messagerequestsrwnode_idrequestbatchesr:r:r;rus                  zSender._send_producer_datac Cs|jr&|jjr&|jr|jjpt}|j||j s&|j |j |jj}|dur4dSt dt||j|jr|jsd}z||rt|j|j}|dur^|j|Wn|jj||jddss|j|d}Wnn%|j}|dur|jj|jd|jdn |jj||jddsWq>|dur|jrt|jdd|j }|j!||j}|"|j#|WdSWn+t$y}zt %d t|||r|j|WYd}~nWYd}~nd}~ww|jr|jrD|dur|j&|dS) NFz$%s: Sending transactional request %srrcr)rdfuturer~Tzs%s: Got an exception when trying to find a node to send a transactional request to. Going to back off and retry: %s)'r- is_completingr$has_incomplete is_abortingrnrNrqrrflush_in_progress begin_flushnext_request_handlerrWrXrYrr'r(needs_coordinator coordinatorcoordinator_typelookup_coordinator_for_requestr# await_readyr least_loaded_noderkr&ris_retryrsleep%next_in_flight_request_correlation_idradd_both on_completer[warningretry)r2r\r target_nodetxn_correlation_idrrvr:r:r;rj$sd             " z"Sender._maybe_send_pending_requestcCs.|jjrtdt|||j|dSdS)Nz4%s: Aborting producer batches due to fatal error: %s)r$rrWerrorrY abort_batches)r2excr:r:r;ro`szSender._maybe_abort_batchescCsd|_|j|dS)zAStart closing the sender (won't complete until all data is sent).FN)r'r$r`rxrar:r:r;initiate_closehs  zSender.initiate_closecCsd|_|dS)z;Closes the sender without sending out any pending messages.TN)r(rrar:r:r; force_closens zSender.force_closecCs0||jjvrt||j||dSdSN)r&_topicsr r*rrx)r2topicr:r:r;ress   zSender.add_topiccCs.tdt||||D]}|||q dS)Nz0%s: Error sending produce request to node %d: %s)rWrrYr)r2rrrrDr:r:r;r~szSender._failed_producec Cstdt|||r3dd|D}|jD]}|j}|jD]}t||j} || } || |qqdSt dd} |D]} || | q:dS)zHandle a produce response.z %s: Parsing produce response: %rcSsi|]}|j|qSr:)r=rzr:r:r; r}z3Sender._handle_produce_response..r) error_codeN) rWrXrY responsesr"partition_responsesr r?_complete_batch_PartitionProduceResponse) r2r send_timerresponsebatches_by_partitiontopic_responserpartition_responserTrD syntheticr:r:r;rs    zSender._handle_produce_responsecs.sfddStfdd}|S)z-Returns a fn mapping batch_index to exceptioncsSrr:_)top_level_exceptionr:r;sz.Sender._record_exceptions_fn..csH|vr tdS|}|ppj}tdkrjntj}||S)Nz^Failed to append record because it was part of a batch which had one more more invalid recordsr)rN KafkaError descriptionrr9InvalidRecordError) batch_index record_errorerr_msgrr record_errorsrecord_errors_dictrr:r;record_exceptions_fnsz:Sender._record_exceptions_fn..record_exceptions_fn)dict)r2rrrrr:rr;_record_exceptions_fns  zSender._record_exceptions_fncCs|j}|dkrt|}|tjurtdt||jd}|dkr[|j}|j }| ||r8| ||j ||jrL|jj|rL|j|j||j|jdrY|j j|jdS|tjurht|jj}n|tjurstd}n||j}||||dS)aWComplete or retry the given batch of records based on a broker response. Handles both the success path (including treating DuplicateSequenceNumberError as success, for max_in_flight > 1 retry arrivals) and the error path, which delegates to _dispatch_error with a context-aware exception instance. Arguments: batch (ProducerBatch): The record batch partition_response (PartitionProduceResponse): Protocol-layer partition response from the broker (or a synthetic instance for the acks=0 case). rz^%s: Received DUPLICATE_SEQUENCE_NUMBER for %s - records already committed, treating as successrNz5The producer is not authorized to do idempotent sends)rrNfor_codeDuplicateSequenceNumberErrorrWrXrYr= base_offsetlog_append_time_mscompleterGr$ deallocater-producer_id_and_epochmatchupdate_last_acked_offsetrr rrTopicAuthorizationFailedErrorrClusterAuthorizationFailedErrorr_dispatch_error)r2rDrr error_clsrlog_append_timer\r:r:r;rs6            zSender._complete_batchcCs&t|tr |d}|j||dddS)aComplete a batch following a client-side failure. Called from _failed_produce for network errors and from _send_producer_data for locally-expired batches. The exception is used as-is (no reconstruction), so any dynamic message is preserved. Arguments: batch (ProducerBatch): The record batch exception (Exception or type): The client-side exception or its class (a bare class is instantiated with no message) N)r) isinstancetyper)r2rDr\r:r:r;rs z%Sender._complete_batch_with_exceptionc Cs2t|}|dur |jnd}|||r9tdt||j|j|j|j || ||j || |n2||||rjtdt||j||j|j|j|j|j || || |n|||r|jr|jddur|jdnd}tdt||j|jd|jd|j|rd |nd td t||j|jr|j|jnd|j || || |n|jr#|jj|||d }|tjkr|jnH|tjkr t|tj r|j!|j"r|dur|j#nd}t$d t||j||j%n|tj&kr|j'|n |tj(kr#|j)||j*r1|j*+|jj,|j|dur=|j+} |j-}nd} |jrP|jddurP|jdnd}|.|| |} |/|| rk| ||j ||tj0ur{tdt||jt1|tj2r|j34|jdr|j j56|jdSdS)aApply the appropriate outcome for a failed batch. Single decision point for both broker-reported errors (with a partition_response) and client-side exceptions (partition_response is None). Handles split / retry / retention-reset / fail along with transaction-state transitions and post-error housekeeping (metadata refresh, partition unmuting). Nr<zN%s: Got %s on topic-partition %s with %d records, splitting batch and retryingz%s: UnknownProducerIdError for %s appears to be retention-based (log_start_offset=%s, last_acked_offset=%s); resetting sequence and retryingrzW%s: Got error produce response on topic-partition %s, retrying (%s attempts left): %s%srrz. Error Message: %sz=%s: Retrying batch to topic-partition %s. Sequence number: %s)log_start_offsetz%s: The broker received an out of order sequence number for topic-partition %s at offset %s. This indicates data loss on the broker, and should be investigated.r:z%s: Received unknown topic or partition error in produce request on partition %s. The topic/partition may not exist or the user may not have Describe access to itr)7rr _can_splitrWrrY__name__r=rr$split_and_reenqueuerGr_record_retries'_is_retention_based_unknown_producer_idr-last_acked_offsetreset_sequence_for_partition reenqueue _can_retryargsr attemptsrXsequence_numberclassify_batch_errorrERROR_CLASS_NEEDS_EPOCH_BUMPbump_producer_id_and_epoch#ERROR_CLASS_NEEDS_PRODUCER_ID_RESETrrNOutOfOrderSequenceNumberErrorrg producer_idrrrERROR_CLASS_FATALtransition_to_fatal_errorERROR_CLASS_ABORTABLEtransition_to_abortable_errorr,rrrrcomplete_exceptionallyUnknownTopicOrPartitionError issubclassInvalidMetadataErrorr&rrr) r2rDr\rrrrclassificationrrrr:r:r;rs          "               &      zSender._dispatch_errorcCs"|jr|j|jj|jdSdSr)r,record_retriesr=rr)r2rDr:r:r;rNszSender._record_retriescCs||jjr dS|j|jdkrdS|jdurdSt|tjs"dS|j rB|j j |sBt dt||j|j|j j j|j j jdSdS)aq We can retry a send if the error is transient, the number of attempts taken is fewer than the maximum allowed, and - for the idempotent producer - the batch's producer id/epoch still matches ours. A mismatched producer id/epoch (e.g. after a reset or future KIP-360 epoch bump) means retrying would violate idempotence. FrNzx%s: Attempted to retry sending a batch but the producer id/epoch changed from %s/%s to %s/%s. This batch will be droppedT)rIr$rJrr rLrrNRetriableErrorr-rrrWrrYr producer_epochepoch)r2rDrr:r:r;rRs   zSender._can_retrycCst|tjurdS|js dS|jj|sdS||jjrdS|jdur%dS|dus-|dkr/dS|j |j }||kS)aDetect retention-based UnknownProducerIdError (KAFKA-5793). The broker returns UnknownProducerIdError either because the producer state was legitimately removed by retention, or because of actual data loss. If the broker's log_start_offset is strictly greater than the last offset we acknowledged for this partition, then the records we previously wrote have been aged out - the producer can safely reset its sequence to 0 and resume. FNr) rNUnknownProducerIdErrorr-rrrIr$rJrLrr=)r2rDrr last_ackedr:r:r;rks  z.Sender._is_retention_based_unknown_producer_idcCs4|tjtjfvo|jdko|jduo||jj S)z We can split and retry a batch if the error indicates the batch is too large for the broker, the batch contains more than one record (so it can actually be split), and the delivery timeout has not been reached. rN)rNMessageSizeTooLargeErrorRecordListTooLargeErrorrrLrIr$rJ)r2rDrr:r:r;rszSender._can_splitcCs>i}|D]\}}|r|||jd|jd|||<q|S)a Transfer the record batches into a list of produce requests on a per-node basis. Arguments: collated: {node_id: [ProducerBatch]} Returns: dict: {node_id: ProduceRequest} (version depends on client api_versions) rr)rH_produce_requestr )r2collatedrrrr:r:r;rs   zSender._create_produce_requestsc sd}d}tjj}tt}|D]} | jj} || jj| j d} ||  | q|j r1|j j nd} | dur9d}t| ||fdd|D||dS) zCreate a produce request from the given record batches. Returns: ProduceRequest (version depends on client api_versions) r)r?recordsNcsg|] \}}||dqS))r"partition_datar:)r{r partitionsTopicr:r;r|sz+Sender._produce_request..)rrrd topic_data min_version max_version)rTopicProduceDataPartitionProduceDatar.r/r0r=r partitionr"bufferrMr-rrH) r2rrtimeoutrr*r) Partitionr(rDrr-rr:r&r;rs2  zSender._produce_requestcCs|jdS)z6Wake up the selector associated with this send thread.N)r#rxrar:r:r;rxsz Sender.wakeupcCs |jSr)r#bootstrap_connectedrar:r:r;r1s zSender.bootstrap_connectedcCsd|jd|jdfS)Nz)rr)r rar:r:r;__str__szSender.__str__r)"r __module__ __qualname____doc__rr rrrGrVrbrZrurjrorrrerrrrrrrrrrrrrxr1r2 __classcell__r:r:r8r;r sP  % /f< 3d !r c@sDeZdZddZ   dddZddZd d Zd d Zd dZdS)r+cs|_|_|_d}j|_jdt|ddjdt|ddd}j|_jdt|d dd }j|_ jd t|d djd t|ddd}j|_ jdt |ddjdt|ddd}j|_ jdt |ddd}j|_ jdt |ddd}j|_jdt |ddd}j|_jdt|ddjdt|d djd!tfd"d#d$d%jd&tfd'd#d(d%dS))Nz batch-sizezbatch-size-avgz;The average number of bytes sent per partition per-request.) sensor_namerzbatch-size-maxz7The max number of bytes sent per partition per-request.compression-ratezcompression-rate-avgz/The average compression rate of record batches.z queue-timezrecord-queue-time-avgzFThe average time in ms record batches spent in the record accumulator.zrecord-queue-time-maxzFThe maximum time in ms record batches spent in the record accumulator.zrecords-per-requestrecord-send-ratez.The average number of records sent per second.zrecords-per-request-avgz*The average number of records per request.bytes byte-ratez,The average number of bytes sent per second.record-retriesrecord-retry-ratez5The average per-second number of retried record sendsrrecord-error-ratezEThe average per-second number of record sends that resulted in errorszrecord-size-maxz*The maximum record size across all batcheszrecord-size-avgz)The average maximum record size per batchzrequests-in-flightcs jSr)r#r^rrar:r;rs z(SenderMetrics.__init__..z=The current number of in-flight requests awaiting a response.)rz metadata-agecs|jjdS)Nr~)r&_last_successful_refresh_ms)rrQrar:r;rsz?The age in seconds of the current producer metadata being used.)rr#r&sensorbatch_size_sensor add_metricrrcompression_rate_sensorqueue_time_sensorrecords_per_request_sensorrbyte_rate_sensor retry_sensor error_sensormax_record_size_sensorr)r2rr3r4r7r:rar;rs             zSenderMetrics.__init__producer-metricsNc CsD|j}|||||}|r||} | ||dS|||dSr)r metric_namer@rrB) r2rK measurable group_namertagsr7mmetricr@r:r:r;rBs  zSenderMetrics.add_metriccsfdd}|j|dse|jdt|dddd|jdt|d dd d|jd t|d dd d|jd t|dddd|jdt|dddddSdS)Ncs d|S)Nz topic.{0}.{1})format)r"rr:r;r7&s z?SenderMetrics.maybe_register_topic_metrics..sensor_namezrecords-per-batchr9zproducer-topic-metrics.z"Records sent per second for topic )r7rMrr;r:zBytes per second for topic r8z$Average Compression ratio for topic r=r<z$Record retries per second for topic r>z record-errorsz#Record errors per second for topic )r get_sensorrBrr)r2rr7r:rRr;maybe_register_topic_metrics$s8       z*SenderMetrics.maybe_register_topic_metricsc Cs|D]}d}d}|D]q}|jj}|||jd|d}||j|jd|d}||j |jd|d} | |j |j |j |j ra|j |j |j|j|j |j|j||j7}||j 7}q |r|j||j|qdS)Nrtopic.z.records-per-batchz.bytesz.compression-rate)rr=rrTrrSrecordrr" size_in_bytescompression_raterAdrainedrDr@rCrImax_record_sizerErF) r2 batches_map node_batchr" total_bytesrDrtopic_records_counttopic_byte_ratetopic_compression_rater:r:r;rFs<         z,SenderMetrics.update_produce_request_metricscC6|j||jd|d}|r||dSdS)NrUz.record-retries)rGrVrrSr2rcountr@r:r:r;rk zSenderMetrics.record_retriescCra)NrUz.record-errors)rHrVrrSrbr:r:r;rqrdzSenderMetrics.record_errors)rJNNN) rr3r4rrBrTrrrr:r:r:r;r+sI  "% r+)#r.rrBlogging threadingrkafkarrNkafka.metrics.measurablerkafka.metrics.statsrrr"kafka.producer.transaction_managerrkafka.protocol.producerrr kafka.structsr kafka.utilr kafka.versionr getLoggerrrWTopicProduceResponsePartitionProduceResponserThreadr r+r:r:r:r;s.       8