o `j[@sddlZddlZddlZddlZddlZddlZddlZddlmZ ddl m Z ddl m Z mZmZmZddlmZddlmZmZmZmZmZddlmZddlmZmZddlmZmZm Z dd l!m"Z"e#e$Z%d a&e'd gd Z(d e(_)e'dgdZ*e'dgdZ+ej,Z-e-j.Z/ej0Z1ej2Z3e3j4Z5ej6Z7e7j8Z9Gddde j:Z;GdddZe>e>j?e>j@e>_Ae>e>j?e>jBe>_CGdddZDGdddZEGdddZFGd d!d!ZGdS)"N)Future)AvgCountMaxRate) FetchRequest)ListOffsetsRequestOffsetForLeaderEpochRequest OffsetSpecUNKNOWN_OFFSETIsolationLevel) MemoryRecords) DeserializerDeserializeWrapper)TopicPartitionOffsetAndMetadataOffsetAndTimestamp)TimerFConsumerRecord) topic partition leader_epochoffset timestamptimestamp_typekeyvalueheaderschecksumserialized_key_sizeserialized_value_sizeserialized_header_sizeaJA single record (message) consumed from a topic partition. Yielded by :meth:`~kafka.KafkaConsumer.poll` (inside the returned ``{TopicPartition: [ConsumerRecord, ...]}`` mapping) and by iterating over a :class:`~kafka.KafkaConsumer`. ``key`` and ``value`` are decoded by the consumer's configured deserializers. Keyword Arguments: topic (str): The topic this record was received from. partition (int): The partition this record was received from. leader_epoch (int): The partition leader epoch for this record, or -1 if unknown. offset (int): The position of this record in the topic partition. timestamp (int): The timestamp of this record, in milliseconds since the epoch (UTC), or -1 if unknown. timestamp_type (int): The type of the timestamp: 0 for CreateTime (set by the producer) or 1 for LogAppendTime (set by the broker). key: The (deserialized) key of the record, or None. value: The (deserialized) value of the record, or None. headers (list): A list of ``(key, value)`` header tuples, where key is a str and value is bytes. checksum (int): Deprecated. The CRC32 checksum of the record, or None. Removed in message format v2 (Kafka 0.11+). serialized_key_size (int): The size of the serialized, uncompressed key in bytes, or -1 if the key is None. serialized_value_size (int): The size of the serialized, uncompressed value in bytes, or -1 if the value is None. serialized_header_size (int): The size of the serialized, uncompressed headers in bytes, or -1 if there are no headers. CompletedFetch)topic_partitionfetched_offsetresponse_versionpartition_datametric_aggregatorExceptionMetadata)rr$ exceptionc@s eZdZdS)RecordTooLargeErrorN)__name__ __module__ __qualname__r.r.T/home/djax/ivt_ai_plugin/venv/lib/python3.10/site-packages/kafka/consumer/fetcher.pyr*Osr*c@seZdZidddddddddd d d d ejd dddddddddddddddddZddZedd Zdmd!d"Z d#d$Z d%d&Z d'd(Z d)d*Z dnd+d,Zdnd-d.Zdnd/d0Zdnd1d2Zdnd3d4Zdnd5d6Zdod7d8Zd9d:Zd;d<Zdnd=d>Zd?d@ZdAdBZdCdDZdEdFZdGdHZdIdJZdndKdLZdndMdNZdOdPZ dQdRZ!dSdTZ"dUdVZ#dWdXZ$dYdZZ%d[d\Z&d]d^Z'd_d`Z(dadbZ)dcddZ*dedfZ+dgdhZ,didjZ-GdkdldlZ.dS)pFetcherkey_deserializerNvalue_deserializerfetch_min_bytesfetch_max_wait_msifetch_max_bytesi max_partition_fetch_bytesimax_poll_records check_crcsTmetricsmetric_group_prefixconsumerrequest_timeout_msi0uretry_backoff_msd!enable_incremental_fetch_sessionsisolation_levelread_uncommitted client_rackmetadata_max_age_msicKsXt|j|_|jD] }||vr|||j|<q dD]&}|j|dur@t|j|ts@tjd|ftddt|j||j|<qz t |jd|_ Wn t yYt ddw||_|j|_|jj|_||_t|_d|_i|_i|_d|_t|_|jdrt|jd|jd |_nd|_i|_t|_d|_ d|_!d|_"d|_#d|_$dS) a_ Initialize a Kafka Message Fetcher. Keyword Arguments: key_deserializer (kafka.serializer.Deserializer): Takes a raw message key and returns a deserialized key. Default: None. value_deserializer (kafka.serializer.Deserializer): Takes a raw message value and returns a deserialized value. Default: None. enable_incremental_fetch_sessions: (bool): Use incremental fetch sessions when available / supported by kafka broker. See KIP-227. Default: True. fetch_min_bytes (int): Minimum amount of data the server should return for a fetch request, otherwise wait up to fetch_max_wait_ms for more data to accumulate. Default: 1. fetch_max_wait_ms (int): The maximum amount of time in milliseconds the server will block before answering the fetch request if there isn't sufficient data to immediately satisfy the requirement given by fetch_min_bytes. Default: 500. fetch_max_bytes (int): The maximum amount of data the server should return for a fetch request. This is not an absolute maximum, if the first message in the first non-empty partition of the fetch is larger than this value, the message will still be returned to ensure that the consumer can make progress. NOTE: consumer performs fetches to multiple brokers in parallel so memory usage will depend on the number of brokers containing partitions for the topic. Supported Kafka version >= 0.10.1.0. Default: 52428800 (50 MB). max_partition_fetch_bytes (int): The maximum amount of data per-partition the server will return. The maximum total memory used for a request = #partitions * max_partition_fetch_bytes. This size must be at least as large as the maximum message size the server allows or else it is possible for the producer to send messages larger than the consumer can fetch. If that happens, the consumer can get stuck trying to fetch a large message on a certain partition. Default: 1048576. check_crcs (bool): Automatically check the CRC32 of the records consumed. This ensures no on-the-wire or on-disk corruption to the messages occurred. This check adds some overhead, so it may be disabled in cases seeking extreme performance. Default: True isolation_level (str): Configure KIP-98 transactional consumer by setting to 'read_committed'. This will cause the consumer to skip records from aborted tranactions. Default: 'read_uncommitted' )r1r2Nz3%s does not implement kafka.serializer.Deserializer)category stacklevelrAzUnrecognized isolation_levelr:r;)%copyDEFAULT_CONFIGconfig isinstancerwarningswarnDeprecationWarningrr build_from_isolation_level ValueErrorErrorsKafkaConfigurationError_client_manager_net_subscriptions collectionsdeque_completed_fetches_next_partition_records_paused_completed_fetches_paused_partition_records _iterator_fetch_futuresFetchManagerMetrics_sensors_session_handlersset"_nodes_with_pending_fetch_requests_cached_list_offsets_exception _next_in_line_exception_metadata _reset_task_validation_task_cached_log_truncation)selfclient subscriptionsconfigsrr.r.r/__init__gsF,        zFetcher.__init__cCs&|jjdus |jjdkrdS|jdS)N)r4r4Fr@)rVbroker_versionrKrkr.r.r/"_enable_incremental_fetch_sessionss z*Fetcher._enable_incremental_fetch_sessionsc s|j||d\}}|s||r|dfSt|j}|jdur)|jjs)||j|s/|dfStfdd}|D]}||q:z |j |j j |Wn t jyXYnw|j||d\}} |dfS)aDrain buffered records, pipeline next fetches, and wait briefly for in-flight responses if no records are immediately available. Single-call replacement for the legacy ``fetched_records -> send_fetches -> client.poll -> fetched_records`` loop in :meth:`KafkaConsumer._poll_once`. The caller no longer drives the event loop; the wait happens inside this method via a wakeup Future fired by any in-flight fetch's completion callback. Arguments: max_records (int, optional): cap on returned records. update_offsets (bool): advance subscription positions for consumed records. timeout_ms (int, optional): wall-clock cap on the wait phase. Only applies when no records are immediately available. Returns: tuple[dict[TopicPartition, list[ConsumerRecord]], bool]: ``(records, idle)``. ``idle`` is True when there were no buffered records, no in-flight fetches, and no pending offset-reset task -- i.e. nothing this fetcher could wait on. Callers in that state should sleep before retrying instead of busy-looping. )update_offsetsFNTcsjs ddSdSN)is_donesuccess)_wakeupr.r/_wakesz$Fetcher.fetch_records.._wake)fetched_records send_fetcheslistr`rhruappendradd_bothrWrunrVwait_forrSKafkaTimeoutError) rk max_recordsrs timeout_msrecordspartial waited_onrzfutrwr.rxr/ fetch_recordss2      zFetcher.fetch_recordscCs|j|jS)zSend FetchRequests for all assigned partitions that do not already have an in-flight fetch or pending fetch data. Returns: List of Futures: each future resolves to a FetchResponse )rVr_send_fetches_asyncrqr.r.r/r| szFetcher.send_fetchescsg}|D]8\}\}}td||j||jj||d}||j ||t | |j |||j|||q |j||IdH|S)NzSending FetchRequest to node %snode_id)_create_fetch_requestsitemslogdebugreaddrVsend add_callback_handle_fetch_responsetime monotonic add_errback_handle_fetch_errorr_clear_pending_fetch_requestr~r`extend_clean_done_fetch_futures)rkfuturesrrequest fetch_offsetsfuturer.r.r/rs    zFetcher._send_fetches_asynccs(|jsdStdd|jD|_dS)Ncss|]}|js|VqdSrtru.0rr.r.r/ Ps  z4Fetcher._clean_done_fetch_futures..)r`rYrZrqr.r.r/r$s * z!Fetcher._clean_done_fetch_futurescCstddt|jDS)zVReturn True if there are any unprocessed (incomplete) FetchRequests in flight.css|]}|j VqdSrtrrr.r.r/r\sz,Fetcher.in_flight_fetches..)anyr}r`rqr.r.r/in_flight_fetchesSs zFetcher.in_flight_fetchescCT|jd}|_|r ||jdur|jjs|jS|jsdS|j|j||_|jS)aeSchedule pending offset resets and return the in-flight Task. Returns the cached Future for the in-flight reset task (shared across concurrent callers) or None if no reset is needed. Callers may discard the Future (fire-and-forget, e.g. consumer.poll) or await it via ``manager.wait_for(future, timeout_ms)`` to block until resets complete (e.g. consumer.position). Arguments: timeout_ms (int, optional): Maximum wall-clock the reset task should run, including time spent awaiting metadata refresh for unknown leaders. If None, uses ``request_timeout_ms`` as a default upper bound so a permanently-unresolvable partition (deleted topic, etc.) doesn't spin forever. The first caller's timeout wins for the cached task; later callers' bounds are enforced via their own ``wait_for`` on the returned Future. Raises: NoOffsetForPartitionError: if a previous reset attempt left a cached non-retriable exception. N)rfrhrurXpartitions_needing_resetrV call_soon_reset_offsets_asyncrkrexcr.r.r/reset_offsets_if_needed^s zFetcher.reset_offsets_if_neededcCs0|j|j||}|D] }||vrd||<q |S)avFetch offset for each partition passed in ``timestamps`` map. Blocks until offsets are obtained, a non-retriable exception is raised or ``timeout_ms`` passed. Arguments: timestamps: {TopicPartition: int} dict with timestamps to fetch offsets by. -1 for the latest available, -2 for the earliest available. Otherwise timestamp is treated as epoch milliseconds. timeout_ms (int, optional): The maximum time in milliseconds to block. Returns: {TopicPartition: OffsetAndTimestamp or None}: Mapping of partition to retrieved offset, timestamp, and leader_epoch. If offset does not exist for the provided timestamp, the value for the TopicPartition will be None. Raises: KafkaTimeoutError if timeout_ms provided N)rWr_fetch_offsets_by_times_async)rk timestampsroffsetstpr.r.r/offsets_by_timess zFetcher.offsets_by_timesc s`siSt|d|f}tt} siS|j|j}zd}d}|j||jIdH\}} Wn'tj yE|jj }Yn*tj y\|jj j rX|jj }nd}Ynw||| sf|Sfdd| D|rz |j||jIdHWn tj yd}Ynw|r|jdd} |jdurt| |jd} |jj| IdH|q) aFetch offsets for each partition in timestamps dict. This may send request to multiple nodes, based on who is Leader for partition. Per-node requests are dispatched concurrently; if any fails, the first exception encountered propagates and the remaining results are dropped. Arguments: timestamps (dict): {TopicPartition: int} mapping of partitions to timestamps or OffsetSpec sentinels. Returns: (fetched_offsets, partitions_to_retry): dict[TopicPartition, OffsetAndTimestamp], set[TopicPartition] Raises: KafkaTimeoutError: if offsets cannot be fully fetched before timeout_ms z,Failed to get offsets by timestamps in %s msTNFcsi|]}||qSr.r.rrrr.r/ sz9Fetcher._fetch_offsets_by_times_async..r>)rrIdictrVr_send_list_offsets_requestsrrrSInvalidMetadataErrorclusterrequest_updateRetriableError need_updateupdaterKminrWsleep maybe_raise) rkrrtimerfetched_offsetsrrefresh_futurebackoffrretrydelayr.rr/rsN    z%Fetcher._fetch_offsets_by_times_asynccC||tj|S)a Fetch earliest (oldest) offset for each partition. Blocks until offsets are obtained, a non-retriable exception is raised or ``timeout_ms`` passed. Arguments: partitions ([TopicPartition]): List of partitions for list offsets. timeout_ms (int, optional): The maximum time in milliseconds to block. Returns: {TopicPartition: int}: Mapping of partition to retrieved offset. Raises: KafkaTimeoutError if timeout_ms provided. )beginning_or_end_offsetr EARLIESTrk partitionsrr.r.r/beginning_offsetszFetcher.beginning_offsetscCr)a Fetch latest (most recent) offset for each partition. Blocks until offsets are obtained, a non-retriable exception is raised or ``timeout_ms`` passed. Arguments: partitions ([TopicPartition]): List of partitions for list offsets. timeout_ms (int, optional): The maximum time in milliseconds to block. Returns: {TopicPartition: int}: Mapping of partition to retrieved offset. Raises: KafkaTimeoutError if timeout_ms provided. )rr LATESTrr.r.r/ end_offsetsrzFetcher.end_offsetscsDtfdd|D}|j|j||}|D] }||j||<q|S)aqFetch offset for each partition using ``timestamp``. Blocks until offsets are obtained, a non-retriable exception is raised or ``timeout_ms`` passed. Arguments: partitions ([TopicPartition]): List of partitions for list offsets. timestamp (int or OffsetSpec): OffsetSpec.LATEST (-1) for the latest available, OffsetSpec.EARLIEST (-2) for the earliest available. Otherwise timestamp is treated as epoch milliseconds. timeout_ms (int, optional): The maximum time in milliseconds to block. Returns: {TopicPartition: int}: Mapping of partition to retrieved offset. Raises: UnsupportedVersionError if broker does not support any compatible ListOffsetsRequest api version. KafkaTimeoutError if timeout_ms provided. csg|]}|fqSr.r.rrr.r/ sz3Fetcher.beginning_or_end_offset..)rrWrrr)rkrrrrrrr.rr/rs zFetcher.beginning_or_end_offsetc Cs|dur |jd}|dkrtd|jdur2|j}d|_|j}|j|r2|j|j|jkr2|j t t }|}d}d}t |j D]}|j|sT|j|j |qB|jdurqt |jD]}|j|sp|j||_nq_zV|dkr|js|js}nI|j} |j| jr| |j | j<qr| j}| j}|| |_n%|jj}|j|r|j|j|<d|_qr|}|jj}||||j||8}|dksvWnty} z|s| t||| |_WYd} ~ nd} ~ wwt|t|jfS)aReturns previously fetched records and updates consumed offsets. Arguments: max_records (int): Maximum number of records returned. Defaults to max_poll_records configuration. Raises: OffsetOutOfRangeError: if no subscription offset_reset_strategy CorruptRecordError: if message crc validation fails (check_crcs must be set to True) RecordTooLargeError: if a message is larger than the currently configured max_partition_fetch_bytes TopicAuthorizationError: if consumer is not authorized to fetch messages from the topic ValueError: if max_records is <= 0 Returns: (records (dict), partial (bool)) records: {TopicPartition: [messages]} partial: True if records returned did not fully drain any pending partition requests. This may be useful for choosing when to pipeline additional fetch requests. Nr8rzmax_records must be > 0)rKrRrgrrX is_fetchablepositionrr$r)rY defaultdictr}r] is_pausedr[r~popr\r^popleftr#_parse_fetched_datanext_fetch_offset_append Exceptionr(rbool) rkrrsexc_metardrainedrecords_remainingfetched_partitionr$ completioner.r.r/r{sr           zFetcher.fetched_recordsc Cs|sdS|j}|j|std|nq|j|s!td|nd|jj|j}|j|j kr{td|j || |}|rD|| ||jj|j }|dur\|j r\|j j||j|s`|swtd||j|jt|jd|j|jj|_t|Std||j|j |dS)NrzMNot returning fetched records for partition %s since it is no longer assignedzWNot returning fetched records for assigned partition %s since it is no longer fetchablez@Returning fetched records at offset %d for assigned partition %szIUpdating fetch position for assigned partition %s to %s (leader epoch %s)rDzMIgnoring fetched records for %s at offset %s since the current position is %d)r#rX is_assignedrrr assignmentrrrtaker highwaterrbrecords_fetch_lagrecordrrlendrain) rkrpartrrsrr part_recordsrr.r.r/rrsH      zFetcher._appendcCs~|j|std|dS|j|std|dS|r/||jj|jks/td|dStd|||j||dS)Nz=Skipping reset of partition %s since it is no longer assignedz>Skipping reset of partition %s since reset is no longer neededzLSkipping reset of partition %s since an alternative reset has been requestedz/Resetting offset for partition %s to offset %s.) rXrrris_offset_reset_neededrreset_strategyinfoseek)rkrrrr.r.r/_reset_offset_if_neededs  zFetcher._reset_offset_if_neededc s|dur |jd}t|}|js|jdurdS|j}|sO|j}|dur*dStd|t }|j dur@t ||j d}|dkrN|j j |IdHqi}|D]}|jj|j}|rb|||<qS|sgdS||} | s|j j} |jd} |j durt | |j } z |j | | IdHWn tjyYnwqtdt|g} | D])\} }t|}t |jdd}|j||| |j |j| ||q| D]}|IdHq|jrdSdS)amDrive resets to completion or until the timer expires. Each iteration fans out per-node ListOffsets requests concurrently and awaits all of them. After a retriable failure (NotLeader, etc.) a partition's next_allowed_retry_time is set ``retry_backoff_ms`` in the future; the loop sleeps until that time and retries rather than relying on an external caller to redrive. If all partitions have unknown leaders, awaits a metadata refresh and retries within the remaining budget. Arguments: timeout_ms (int, optional): Hard upper bound on the loop's wall-clock. None falls back to ``request_timeout_ms`` so a deleted-topic / permanently-unknown-leader partition can't spin the loop forever. The metadata-refresh wait inside the loop is capped by ``min(remaining_timer, request_timeout_ms)``. Per-node failures are caught inside _reset_offsets_for_node and stuffed into self._cached_list_offsets_exception; the next call to reset_offsets_if_needed surfaces them. Nr=rrzResetting offsets for %s) rKrexpiredrfrXrnext_offset_reset_retry_timemaxrrrrrVrWrrr_group_list_offset_requestsrrrrSrrrrdkeysrset_reset_pendingr~r_reset_offsets_for_node)rkrrr next_retryr offset_resetsrtstimestamps_by_nodemetadata_updatewait_ms node_tasksrt_and_enode_partitions expire_attaskr.r.r/rsd              zFetcher._reset_offsets_asyncc sz |||IdH\}}Wn<tyJ}z0|j|t|jdd|jj t |t j s?|j s9||_ ntd|WYd}~dSd}~ww|rc|j|t|jdd|jj |D]\}}||\} } ||| |jqgdS)Nr>rzKDiscarding error in ListOffsetResponse because another error is pending: %s)_send_list_offsets_requestrrX reset_failedrrrKrVrrrLrSrrfrerrorrrr) rkrtimestamps_and_epochsrrpartitions_to_retryrrrr_epochr.r.r/rs(       zFetcher._reset_offsets_for_nodec sn|}|s tfdd|D}t}t}|D]}|IdH\}}||||q||fS)aFetch offsets for each partition in timestamps dict. This may send request to multiple nodes, based on who is Leader for partition. Per-node requests are dispatched concurrently; if any fails, the first exception encountered propagates and the remaining results are dropped. Arguments: timestamps (dict): {TopicPartition: int} mapping of fetching timestamps. Returns: (fetched_offsets, partitions_to_retry): dict[TopicPartition, OffsetAndTimestamp], set[TopicPartition] Raises: StaleMetadata: if no node has known leader for any partition. cs"g|] \}}jj||qSr.)rVrr)rrrrqr.r/r*sz7Fetcher._send_list_offsets_requests..N)rrS StaleMetadatarrrdr) rkrrrrr foffsrr.rqr/rs    z#Fetcher._send_list_offsets_requestscCstt}|D]?\}}|jj|}|dur-|jj|jt d||jj q |dkr>t d||jj q d}||f|||<q t|S)Nz+Partition %s is unknown for fetching offsetrzRLeader for partition %s unavailable for fetching offset, wait for metadata refresh) rYrrrrVrleader_for_partition add_topicrrrr)rkrrrrrrr.r.r/r7s  z#Fetcher._group_list_offset_requestsc stdd|Drdnd}t|t|j}tt}| D]\}\}}t |j ||d}||j  |q"t|jt| |d} td| ||jj| |dId H} || S) a7Send single ListOffsetsResponse to node_id Returns: (fetched_offsets, partitions_to_retry): dict[TopicPartition, OffsetAndTimestamp], set[TopicPartition] Raises: TopicAuthorizationFailedError: if any topic returned an auth error css|] }|ddkVqdS)rNr.)rresr.r.r/rSsz5Fetcher._send_list_offsets_request..r4r)partition_indexcurrent_leader_epochr)rAtopics min_versionz)Sending ListOffsetRequest %s to broker %srN)rvaluesrrmin_version_for_isolation_levelrQrYrr}r_ListOffsetsPartitionrrr~rrrVr_handle_list_offsets_response) rkrr rby_topicrrrdatarresponser.r.r/rHs&    z"Fetcher._send_list_offsets_requestc Cst}t}t}|jD]}|jD]}t|j|j}|j}t |} | tj urm|j dkrB|j } t | dkr9td| r?| dnt} n|j} |j} |j} | durQd} | durWd} td|| | | | tkrlt| | | ||<q| tjurytd|q| tjtjtjtjtjfvrtd| j|||q| tjurtd |||q| tj ur||j!qtd || j||qq |rt |||fS) atParse a ListOffsets response. Returns: (fetched_offsets, partitions_to_retry): dict[TopicPartition, OffsetAndTimestamp], set[TopicPartition] Raises: TopicAuthorizationFailedError: if any topic returned an auth error ValueError: if ListOffsetsResponse v0 and > 1 offset returned rr4z,Expected ListOffsetsResponse with one offsetNrz^Handling ListOffsetsResponse response for %s. Fetched offset %s, timestamp %s, leader_epoch %sz_Cannot search by timestamp for partition %s because the message format version is before 0.10.0zEAttempt to fetch offsets for partition %s failed due to %s, retrying.zReceived unknown topic or partition error in ListOffsets request for partition %s. The topic/partition may not exist or the user may not have Describe access to it.z;Attempt to fetch offsets for partition %s failed due to: %s)"rrdrrrnamer error_coderSfor_codeNoError API_VERSIONold_style_offsetsrrRr rrrrrr UnsupportedForMessageFormatErrorNotLeaderForPartitionErrorReplicaNotAvailableErrorKafkaStorageErrorOffsetNotAvailableErrorLeaderNotAvailableErrorr+rUnknownTopicOrPartitionErrorwarningTopicAuthorizationFailedErrorr)rkrrr unauthorized_topics topic_datapartition_inforr error_typerrrrr.r.r/rgsp             3 z%Fetcher._handle_list_offsets_responsecCs0|jD]}|jj|}|j||qdS)aWalk assigned partitions; mark any whose cluster leader epoch has advanced beyond the position's epoch as awaiting validation. Cheap fire-and-forget marker; the actual RPC fan-out runs in ``validate_offsets_if_needed`` -> ``_validate_offsets_async``. Idempotent: partitions already awaiting validation, awaiting reset, or with no recorded epoch are skipped inside ``maybe_validate_position``. N)rXassigned_partitionsrVrleader_epoch_for_partition*maybe_validate_position_for_current_leader)rkr current_epochr.r.r/maybe_validate_positionss z Fetcher.maybe_validate_positionscCr)aLSchedule any pending position validations and return the in-flight Task. Mirrors :meth:`reset_offsets_if_needed`: returns a cached Future shared across callers so concurrent ``consumer.poll`` and ``consumer.position`` callers don't race the same partition into duplicate OffsetForLeaderEpoch requests. Raises: LogTruncationError: if a previous validation detected truncation on one or more partitions. The exception is cleared after being raised so subsequent calls will re-attempt validation. N)rjrirurXpartitions_needing_validationrVr_validate_offsets_asyncrr.r.r/validate_offsets_if_neededs  z"Fetcher.validate_offsets_if_neededcs|dur |jd}t|}|js|jdurdS|j}|sO|j}|dur*dStd|t }|j dur@t ||j d}|dkrN|j j |IdHqi}|D]}|jj|}|jdurk|jjdkrk|j||<qS|spdS||} | s|j j} |jd} |j durt | |j } z |j | | IdHWn tjyYnwqtdt|g} | D](\} }t|}t |jdd}|j||| |j |j | |q| D]}|IdHq|jrdSdS)aDrive offset validations to completion or until the timer expires. Same overall shape as ``_reset_offsets_async``: per-node fan-out. After a retriable failure (FencedLeaderEpoch, etc.) a partition's next_allowed_retry_time is set ``retry_backoff_ms`` in the future; the loop sleeps until that time and retries rather than relying on an external caller to redrive. Stops on first ``LogTruncationError`` accumulation; the next caller surfaces it. Nr=rrrzValidating offsets for %s)!rKrrrjrXr5!next_offset_validation_retry_timerrrrrrVrWrrrr'_group_offset_for_leader_epoch_requestsrrrrSrrrrdrrset_validation_pendingr~r_validate_offsets_for_node)rkrrrrr positionsrstaterequests_by_noderrrrpayloadrrrr.r.r/r6sd               zFetcher._validate_offsets_asyncc sz |||IdH}Wn8tyD}z,|jt|t|jdd|jj t |t j s9td||WYd}~dSd}~ww|r]|jdurTt ||_dS|jj|dSdS)Nr>rzrNrzNSkipping validation completion for %s: position changed since request was sentrzTruncation detected for %s at position %s (broker returned UNDEFINED end_offset/leader_epoch); resetting offset per auto_offset_reset policyzyTruncation detected for %s at position %s (broker returned UNDEFINED end_offset/leader_epoch), but no reset policy is setrDzOTruncation detected for %s at position %s; seeking to first diverging offset %szdTruncation detected for %s at position %s (first diverging offset is %s), but no reset policy is setzKOffsetForLeaderEpoch for %s returned retriable %s; will retry after backoffz4OffsetForLeaderEpoch for %s: unknown topic/partitionz*OffsetForLeaderEpoch for %s failed with %s)-rdrrrKrrrrrgetrSrrr  end_offsetrrXrrrrrhas_default_offset_reset_policyrrrequest_offset_resetr*complete_validationrrmetadataFencedLeaderEpochErrorUnknownLeaderEpochErrorr$r%r&r(r+rVrrrr)r+rA)rkrrequested_positionsrEr,retry_atrr-r.r requestedr/rK end_epochcurrenthas_reset_policy divergent validatedr.r.r/rI`s                  R z0Fetcher._handle_offset_for_leader_epoch_responsecs^|j}dd|jD|j}|r|j|j|j fdd|DS)NcSsh|]}|jqSr.)r#)rfetchr.r.r/ sz0Fetcher._fetchable_partitions..csg|]}|vr|qSr.r.rdiscardr.r/rz1Fetcher._fetchable_partitions..) rXfetchable_partitionsr[rIr\rr#rr]r^)rk fetchablerVr.r\r/_fetchable_partitionss    zFetcher._fetchable_partitionscCsl|jj|}|dur|jj|S|jj||s4|jj||jj|}t d||||S|S)aPick the node to fetch from for ``tp``: a cached preferred read replica (KIP-392) when valid and *still listed as a replica of ``tp``*, otherwise the partition leader. A preferred replica that has been demoted out of the partition's replica set (or fell out of cluster metadata entirely) is cleared so the next fetch goes to the leader. NzmPreferred read replica %s for partition %s no longer online or no longer a replica; falling back to leader %s) rXrpreferred_read_replicarVrris_replica_nodeclear_preferred_read_replicarr)rkr preferredleaderr.r.r/_select_read_replicaszFetcher._select_read_replicac sd}ttj}|D]z}||}|jj|j}|dus"|dkr/t d||j j q |j |dkr?t d||q |j|dkrOt d||q ||jvr\t d||q |j j |}|durid}t|j||j|j|jd d }||||<t d ||jq i}|D]X\}|jr||jvrt||j|<|j|} ntdtj} t |j!} t|jd |jd |jd|j!| j"| j#| j$| j%|jd| |d } fddD} | | f||<q|S)zCreate fetch requests for all assigned partitions, grouped by node. FetchRequests skipped if no leader, or node has requests in flight Returns: dict: {node_id: (FetchRequest, {TopicPartition: fetch_offset}), ...} Nrz.)&rYr OrderedDictrargrXrrrrrVrrconnection_delayrUthrottle_delayrer1_FetchPartitionrrrrKrrrrcFetchSessionHandler build_nextFetchRequestData FetchMetadataLEGACYrrrQidepochto_send to_forget) rkrsr`rrrrr.requestssessionrrrr.rtr/rsx        zFetcher._create_fetch_requestsc Cs|jdkr|jr||jvrtd|dS|j||sdStdd|jD}|jr3t |j|}nd}|jD]"}|j D]}t |j |j } || } t| | |j||} |j| q=q8|jrm|jjt|ddSdS)z!The callback for fetch completionzIUnable to find fetch session handler for node %s. Ignoring fetch responseNcSs&g|]}|jD] }t|j|jqqSr.rrrrrr-r&r.r.r/rPsz2Fetcher._handle_fetch_response..r)r!rrrcrrhandle_responserd responsesrbFetchResponseMetricAggregatorrrrrr"r[r~ fetch_latencyrrr) rkrr send_timerrr'r-r&rricompleted_fetchr.r.r/rGs<    zFetcher._handle_fetch_responsecCsJt|tjr tjntj}t|d||||jvr#|j||dSdS)NzFetch to node %s failed: %s) rLrS CancelledloggingINFOERRORrrc handle_error)rkrr)levelr.r.r/rns  zFetcher._handle_fetch_errorcCs(z |j|WdStyYdSwrt)reremoveKeyError)rkrrwr.r.r/rts  z$Fetcher._clear_pending_fetch_requestcCsr|j}|dus |jdkrdS|jj||j|jr5td||j|j|jj|jdur7|jj dSdSdS)aUApply a KIP-951 ``current_leader`` hint from a Fetch v12+ response. Updates the cluster's cached leader id/epoch when the broker advertises a newer leader. If the new leader id is not yet a known broker (v12 has no ``node_endpoints``), requests a metadata refresh so the consumer learns its address. Nrz=Fetch response advertised new leader for %s: node %s epoch %s) current_leaderrrVrupdate_partition_leader leader_idrrbroker_metadatar)rkrr&rfr.r.r/_maybe_update_current_leaderzs  z$Fetcher._maybe_update_current_leaderc Cs|j}|j}|jj}|jj}t|}d}z|j|s%t d|n|tj ur9|jj |j }|dus;|j|krct d|||jW|durT|jrT|j|dd|tj ura|j|dSdS|jj} | dur| jdkrt d|| j| j|j||jjW|dur|jr|j|dd|tj ur|j|dSdSt|jj} |jj} t d| |||j||| |jd|jd|jd|j | |j|j!d }| "s| dkr|j#d kr||i} t$d | |jd f| t%d ||f|dkr||jj |_&|jj'} |jj |(| t)*|jddr8| dus*| dkr1t d|nt d|| n|tj+tj,tj-tj.fvr\t d||j/|0||j|jjn|tj1tj2fvrt d||j/|0||j|j||jjn|tj3ur|jj |j }|dus|j|krt d|||jnh|jj |4}|durt d|||nR|j5rt d|||j6|n>t3||i|tj7urt 8d|j9t7t:|j9gt;|tj<rt d||t;|tj=r|jjn|dW|dur|jr|j|dd|tj ur)|j||S|dur<|jr<|j|dd|tj urI|j|ww)NzIIgnoring fetched records for partition %s since it is no longer fetchablezdDiscarding fetch response for partition %s since its offset %d does not match the expected offset %drzLFetch for %s diverged at epoch %s offset %s; marking position for validationzBPreparing to read %s bytes of data for partition %s with offset %dr1r2r9)r1r2r9rAaborted_transactionsr'on_drainrFa'There are some messages at [Partition=Offset]: %s whose size is larger than the fetch size %s and hence cannot be ever returned. Please condier upgrading your broker to 0.10.1.0 or newer to avoid this issue. Alternatively, increase the fetch size on the client (using max_partition_fetch_bytes)r7zFailed to make progress reading messages at %s=%s. Received a non-empty fetch response from the server, but no complete records were found.rEg@@z/Cleared preferred read replica for partition %sz6Updating preferred read replica for partition %s to %szError fetching partition %s: %sz9Fetch for %s returned %s; marking position for validationzqDiscarding stale fetch response for partition %s since the fetched offset %d does not match the current offset %dzHFetch offset %s out of range for %s on follower %s; retrying from leaderz6Fetch offset %s is out of range for topic-partition %sz%Not authorized to read from topic %s.z)Retriable error fetching partition %s: %sz$Unexpected error while fetching data)>r#r$r&rhigh_watermarkrSrrXrrrr rrrr'rmove_partition_to_enddiverging_epochrKrrrequest_position_validationrVrrr rr size_in_bytesPartitionRecordsrKrQ_on_partition_records_drainhas_nextr%r* KafkaErrorrrbupdate_preferred_read_replicarrr$r%r)r&r+rrPrQOffsetOutOfRangeErrorrdrLrMr+r*rrd issubclassrr)rkrrrirrr/parsed_recordsrrrrrecord_too_large_partitionsrbclearedr.r.r/rs   r    e                 zFetcher._parse_fetched_datacCs |jdkr|j|jdSdSNr) bytes_readrXrr#)rkpartition_recordsr.r.r/r&s z#Fetcher._on_partition_records_draincCsJ|jdur |j|jD]}|q|j|jd|_dSrt)r\rr^rclearr]rg)rkparkedr.r.r/close0s      z Fetcher.closec@s|eZdZdddejddddfddZddZd d Zd d Zd dZ dddZ ddZ ddZ ddZ ddZddZdS)zFetcher.PartitionRecordsNTcCsdSrtr.)xr.r.r/?sz!Fetcher.PartitionRecords.c Cs||_||_d|_||_d|_d|_||_t|_t t |pgddd|_ | |_ ||_t|j||||||_| |_d|_dS)NrrcS|jSrt) first_offset)txnr.r.r/rIsz3Fetcher.PartitionRecords.__init__..)r)rir#rrr records_readrArdaborted_producer_idsrYrZsortedrr'r9 itertools dropwhile_maybe_skip_record_unpack_recordsrecord_iteratorr_next_inline_exception) rkrirrr1r2r9rArr'rr.r.r/ro:s& z!Fetcher.PartitionRecords.__init__cCs&|j|jkrtd|j|jdSdS)Nz*Skipping message offset: %s (expecting %s)TF)rrirr)rkrr.r.r/rSs z+Fetcher.PartitionRecords._maybe_skip_recordcCs |jduSrt)rrqr.r.r/__bool__`s z!Fetcher.PartitionRecords.__bool__cCsD|jdur d|_d|_|jr|j|j|j|j||dSdSrt)rrr'rr#rrrrqr.r.r/rcs zFetcher.PartitionRecords.draincCs|jr |jd}|_|dSrt)r)rkrr.r.r/"_maybe_raise_next_inline_exceptionksz;Fetcher.PartitionRecords._maybe_raise_next_inline_exceptionc Cs^|g}z|t|jd|W|Sty.}z|s |||_WYd}~|Sd}~wwr)rrrislicerrr)rknrrr.r.r/rpszFetcher.PartitionRecords.takeccsVz|}d}|dur|}|jr"|s"td|j|jf|jdkr{|j|_|j t j kro| ro| |j|j}||rUz|j|Wn#tyTYnw||rotd|j||j|j|j|_|}q |jr{|j|_|}q |D]}|jr|std|j|jf|jdurt|jnd} |jdurt|jnd} |||j|j |j} |||j|j |j} |j } | rt!dd| Dnd}|j"d7_"|j#|j$7_#|jd|_t%|j|j&|j|j|j'|j(| | |j |j)| | | Vq}|}|dus|r|jdkr|j|_|*WdSt+y*t,d t-d w) Nz;Record batch for partition %s at offset %s failed crc checkzXSkipping aborted record batch from partition %s with producer_id %s and offsets %s to %sz5Record for partition %s at offset %s failed crc checkrcss6|]\}}t|d|durt|ndVqdS)zutf-8Nr)rencode)rh_keyh_valr.r.r/rs& z;Fetcher.PartitionRecords._unpack_records..r4z)StopIteration raised unpacking messageset). next_batchr9 validate_crcrSCorruptRecordErrorr# base_offsetmagicrrAr READ_COMMITTEDhas_producer_id#_consume_aborted_transactions_up_to last_offset producer_id_contains_abort_markerrrr_is_batch_abortedrr next_offsetris_control_batchrrrr _deserializerrsumrrrrrrrrr StopIterationr) RuntimeError)rkrrr1r2batch last_batchrrkey_size value_sizerrr header_sizer.r.r/r}s        D z(Fetcher.PartitionRecords._unpack_recordscCsR|dur|Sz||||WSty(ts tjdtdd}|||YSw)NzAdeserializer does not implement deserialize(topic, headers, data))rGT) deserialize TypeError_LOGGED_DESERIALIZE_WARNINGrMrNrO)rk deserializerrrrLOGGED_DESERIALIZE_WARNINGr.r.r/rs z%Fetcher.PartitionRecords._deserializecCsZ|jsdS|jr'|jdj|kr+|j|jj|jr)|jdj|ksdSdSdSdSr)rrrrrr)rkrr.r.r/rs &z TopicPartitionState Returns: FetchRequestData z5Built full fetch %s for node %s with %s partition(s).Nz;Building incremental partitions from next: %s, previous: %szRBuilt incremental fetch %s for node %s. Added %s, altered %s, removed %s out of %scs"i|] }|Bvr||qSr.r.raddedalteredrur.r/r!s"z2FetchSessionHandler.build_next..)ris_fullrrrrrr|rdrrrrYrv)rkruprev_tpsnext_tpsrremovedrr.rr/r{s4   zFetchSessionHandler.build_nextc Cs|jtjjkr+t|j}td|j|j||tj ur#t j |_dS|j |_dS| |}t|j}|jjr||krRtd|j||||t j |_dS|jt jkrhtd|jt|t j |_dS|jt jkrztd|jt|dStd|j|jt|t |j|_dS||rtd|j|||j |_dS|jt jkrtd |j|jjt|t|jt|t j |_dS|jt jkrtd |jt|dStd |j|jt|t|jt||j|_dS) Nz.)r)rkrr.r.r/rhsz(FetchSessionHandler._response_partitionsN) r+r,r-__doc__ror{rrrr.r.r.r/rzs "A rzc@s`eZdZdZdZdZdZdZdZddZ e ddZ e d d Z d d Ze d dZddZdS)r}rorirrcCs||_||_dSrtr)rkrorr.r.r/row zFetchMetadata.__init__cCs|j|jkp |j|jkSrt)r INITIAL_EPOCH FINAL_EPOCHrqr.r.r/r{szFetchMetadata.is_fullcCs$|dkr|jS||jkrdS|dS)Nrr4)r MAX_EPOCH)cls prev_epochr.r.r/ next_epochs  zFetchMetadata.next_epochcCs||j|jSrt) __class__rorrqr.r.r/rrz!FetchMetadata.next_close_existingcCs||||jSrt)r r)rror.r.r/rszFetchMetadata.new_incrementalcCs||j||jSrt)r ror rrqr.r.r/rszFetchMetadata.next_incrementalN)r+r,r- __slots__rrrrrrorr classmethodr rrrr.r.r.r/r}ns     r}c@sTeZdZdZddZeddZeddZedd Zed d Z ed d Z dS)r|)_to_send _to_forget _metadatacCs"|pt|_|p t|_||_dSrt)rr rdrr)rkrrrOr.r.r/ros   zFetchRequestData.__init__cCrrt)rrqr.r.r/rOszFetchRequestData.metadatacC|jjSrt)rrorqr.r.r/rzFetchRequestData.idcCrrt)rrrqr.r.r/rrzFetchRequestData.epochcCs@tt}|jD] \}}||j|q dd|DS)NcSg|] \}}t||dqS)rr) _FetchTopicrrrr.r.r/r z,FetchRequestData.to_send..)rYrr}r rrr~)rkr&rr.r.r.r/rs zFetchRequestData.to_sendcCs:tt}|jD] }||j|jqdd|DS)NcSrr)_ForgottenTopicrr.r.r/rrz.FetchRequestData.to_forget..)rYrr}rrr~rr)rkr&rr.r.r/rs  zFetchRequestData.to_forgetN) r+r,r-r rorrOrrrrr.r.r.r/r|s     r|c@seZdZdZddZdS) FetchMetrics total_bytes total_recordscCsd|_d|_dSrrrqr.r.r/rorzFetchMetrics.__init__N)r+r,r-r ror.r.r.r/rs rc@s eZdZdZddZddZdS)ra Since we parse the message data for each partition from each fetch response lazily, fetch-level metrics need to be aggregated as the messages from each partition are parsed. This class is used to facilitate this incremental aggregation. cCs$||_||_t|_tt|_dSrt)sensorsunrecorded_partitionsr fetch_metricsrYrtopic_fetch_metrics)rkrrr.r.r/rosz&FetchResponseMetricAggregator.__init__cCs|j||jj|7_|jj|7_|j|jj|7_|j|jj|7_|jsU|jj |jj|jj |jj|j D]\}}|j ||j|jqFdSdS)z After each partition is parsed, we update the current metric totals with the total bytes and number of records parsed. After all partitions have reported, we write the metric. N) rrrrrrrr bytes_fetchedrrecords_fetchedrrecord_topic_fetch_metrics)rkr num_bytes num_recordsrr:r.r.r/rs z$FetchResponseMetricAggregator.recordN)r+r,r-rrorr.r.r.r/rs rc@seZdZddZddZdS)racCsJ||_d|f|_|d|_|j|d|jdt|j|d|jdt|j|d|jdt|jd |_ |j |d |jd t|j |d |jd t|d|_ |j |d|jdt|j |d|jdt|j |d|jdtt d|d|_ |j |d|jdtdS)Nz%s-fetch-manager-metrics bytes-fetchedfetch-size-avgz/The average number of bytes fetched per requestfetch-size-maxz/The maximum number of bytes fetched per requestbytes-consumed-ratez/The average number of bytes consumed per secondrecords-fetchedrecords-per-request-avgz-The average number of records in each requestrecords-consumed-ratez1The average number of records consumed per secondz fetch-latencyzfetch-latency-avgz+The average time taken for a fetch request.zfetch-latency-maxz)The max time taken for any fetch request.z fetch-ratez(The number of fetch requests per second.) sampled_statz records-lagzrecords-lag-maxzNThe maximum lag in terms of number of records for any partition in self window) r: group_namesensorr r metric_namerrrr!rrr)rkr:prefixr.r.r/rosf     zFetchManagerMetrics.__init__cCsDdd|dg}|j|}|sTd|ddi}|j|}||jd|jd|f|t||jd|jd|f|t ||jd |jd |f|t | |dd|d g}|j|}|sd|ddi}|j|}||jd |jd |f|t||jd|jd|f|t | |dS)N.rr%rwr&zThe average number of records consumed per second for topic %s) joinr: get_sensorreplacer.rr/r-rrrr)rkrr#r$rr  metric_tagsr!r.r.r/r" s^          z.FetchManagerMetrics.record_topic_fetch_metricsN)r+r,r-ror"r.r.r.r/ras ra)HrYrIrrrrrM kafka.errorserrorsrS kafka.futurerkafka.metrics.statsrrrrkafka.protocol.consumerrrr r r r kafka.recordr kafka.serializerrr kafka.structsrrr kafka.utilr getLoggerr+rr namedtuplerrr"r( FetchTopicrFetchPartitionryForgottenTopicrListOffsetsTopic_ListOffsetsTopicListOffsetsPartitionrOffsetForLeaderTopic_OffsetForLeaderTopicOffsetForLeaderPartitionrHrr*r0rzr}rrrrr~r|rrrar.r.r.r/st       )$-!