Ii@SSKJrJr SSKrSSKrSSKrSSKrSSKrSSKrSSK J r SSK J r SSKJr SSKJrJrJrJr SSKJrJr SSKJrJrJr SSKJr SS KJr SS K J!r!J"r"J#r# SS K$J%r% \RL"\'5r(Sr)S r*\)\*S .r+\RX"S/SQ5r-\RX"S/SQ5r.\RX"S/SQ5r/"SS\ R`5r1"SS\ R`5r2"SS\ Rf5r4"SS\55r6"SS\55r7\7"\7Rp\7Rr5\7l:\7"\7Rp\7Rv5\7l<"SS\55r="S S!\55r>"S"S#\55r?"S$S%\55r@g)&)absolute_importdivisionN)six)Future)AvgCountMaxRate) FetchRequestAbortedTransaction)ListOffsetsRequestOffsetResetStrategyUNKNOWN_OFFSET) MemoryRecords) Deserializer)TopicPartitionOffsetAndMetadataOffsetAndTimestamp)Timer)read_uncommittedread_committedConsumerRecord) topic partition leader_epochoffset timestamptimestamp_typekeyvalueheaderschecksumserialized_key_sizeserialized_value_sizeserialized_header_sizeCompletedFetch)topic_partitionfetched_offsetresponse_versionpartition_datametric_aggregatorExceptionMetadata)rr) exceptionc\rSrSrSrg)NoOffsetForPartitionError2N__name__ __module__ __qualname____firstlineno____static_attributes__r2= 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' rQzUnrecognized isolation_levelNrLrM)copyDEFAULT_CONFIGconfigISOLATION_LEVEL_CONFIGErrorsKafkaConfigurationError_client_subscriptions collectionsdeque_completed_fetches_next_partition_records _iterator_fetch_futuresFetchManagerMetrics_sensors_isolation_level_session_handlersset"_nodes_with_pending_fetch_requests_cached_list_offsets_exception _next_in_line_exception_metadata)selfclient subscriptionsconfigsr s r:__init__Fetcher.__init__LsTii 3 34 ;;C~#*< C  ;;( )1G G001OP P +"-"3"3"5'+$)//1 ;;y !/ I0F TiHjkDM DM 6t{{CT7U V!#25%/.2+04-r9cH/n[R"UR55Hunup4[R SU5 UR R U5 URRX#SS9nURURX$[R"55 URURU5 URURU5 UR!U5 M UR"R%U5 UR'5 U$)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 zSending FetchRequest to node %sF)wakeup)r iteritems_create_fetch_requestslogdebugrfaddrYsend add_callback_handle_fetch_responsetime add_errback_handle_fetch_erroradd_both_clear_pending_fetch_requestappendr`extend_clean_done_fetch_futures)rifuturesnode_idrequest fetch_offsetsfutures r: send_fetchesFetcher.send_fetchess14t?Z?Z?\1] -G-g II7 A  3 3 7 7 @\\&&w&FF    ; ;WUYU^U^U` a   t77 A OOD==w G NN6 "2^ ""7+ &&(r9cUR(dgURSR(dgURR5 MMNr)r`is_donepopleftris r:r!Fetcher._clean_done_fetch_futuress?&&&&q)11    ' ' ) r9cLUR5 [UR5$)zAReturn True if there are any unprocessed FetchRequests in flight.)rboolr`rs r:in_flight_fetchesFetcher.in_flight_fetchess &&(D''((r9cTURSsolU(aUeURR5nU(dg[R SU5 [ 5nUH3nURR URnU(dM/XSU'M5 URU5 g)atReset offsets for the given partitions using the offset reset strategy. Arguments: partitions ([TopicPartition]): the partitions that need offsets reset Returns: bool: True if any partitions need reset; otherwise False (no reset pending) Raises: NoOffsetForPartitionError: if no offset reset strategy is defined KafkaTimeoutError if timeout_ms provided NFzResetting offsets for %sT) rgrZpartitions_needing_resetrsrtdict assignmentreset_strategy_reset_offsets_async)riexc partitions offset_resetstptss r:reset_offsets_if_neededFetcher.reset_offsets_if_neededs483V3VX\0 0 I((AAC  ,j9 B$$//3BBBr$&b! !!-0r9cPURX5nUHnXC;dM SX4'M U$)aFetch 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}: Mapping of partition to retrieved offset, timestamp, and leader_epoch. If offset does not exist for the provided timestamp, that partition will be missing from this mapping. Raises: KafkaTimeoutError if timeout_ms provided N)_fetch_offsets_by_times)ri timestamps timeout_msoffsetsrs r:offsets_by_timesFetcher.offsets_by_timess2*..zFB " r9cU(d0$[USU<S35n[R"U5n[5nU(d0$URU5nURR XSR S9 UR(dGOUR5(aVURURS5 URS(dU$URSVs0sHofX_M nnO!UR5(d UReURR(d%URRR(aZURRR!5nURR XsR S9 UR(dOOtUR bUR UR"S:a'[$R&"UR"SS- 5 O#[$R&"UR S- 5 UR)5 GM[*R,"SU<S35es snf)Nz'Failed to get offsets by timestamps in z ms)rrrrrO)rrSr_send_list_offsets_requestsrYpollrr succeededupdater! retriabler.invalid_metadatacluster need_updaterequest_updaterUrysleep maybe_raiserWKafkaTimeoutError)rirrtimerfetched_offsetsrrrefresh_futures r:rFetcher._fetch_offsets_by_timessIjT^"`aYYz* & 55jAF LL  V8H8H  I>>!!&&v||A7||A**;A<<?K?R*.0? K %%''&&&00DLL4H4H4T4T!%!5!5!D!D!F !!DTDT!U~~&##+u/?/?$++N`Ba/aJJt{{+=>EFJJu//$67    CF&&>H JL L'Ls&IcDURU[RU5$N)beginning_or_end_offsetrEARLIESTrirrs r:beginning_offsetsFetcher.beginning_offsetss%++ +44jB Br9cDURU[RU5$r)rrLATESTrs r: end_offsetsFetcher.end_offsetss%++ +22J@ @r9c[UVs/sHoDU4PM sn5nURXS5nUHnXdRXd'M U$s snfr)rrr)rirrrrrrs r:rFetcher.beginning_or_end_offsetsQZ@Zr ?Z@A ..zFB!+,,GK As A cUcURSnUS:deURb~URnSUlURnURR U5(a?URR U5R UR:Xa URe[R"[5nUnSnSnUS:aUR(d[UR(dOURR5n U RnU RnUR!U 5Ul OLURRnURR"nX`R%UURUU5-nUS:aM[+U5[-UR54$![&a$n U(dU e[)XxU 5UlSn A NISn A ff=f)aIReturns 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 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. NrJr)rUrhrrZ is_fetchablepositionrr)r.r[ defaultdictlistr^r]rr(_parse_fetched_datanext_fetch_offset_append Exceptionr-rr) ri max_recordsupdate_offsetsexc_metardrainedrecords_remainingfetched_partitionr) completiones r:fetched_recordsFetcher.fetched_records&s,  ++&89KQ  0 0 <<%>N373K3KJ3WD0(,(D(D(T(T%%)%A%A%S%SN%g6:6R6R6G6D*FF%$a'(G}d4#:#:;;;  l4EFWij4kD 1 1  ls(F-0BF-- G7GGcTU(dgURnURRU5(d[R SU5 GOURR U5(d[R SU5 GOURR URnURUR:XGa+[R SURU5 URU5nU(aXRU5 URR URnUbBUR(a1URRRXR- 5 U(dU(d^[R SXRRUR 5 [#URSS5URR Ul[%U5$[R SXRRUR5 UR'5 g) 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)rzMIgnoring fetched records for %s at offset %s since the current position is %d)r(rZ is_assignedrsrtrrrrrtaker highwaterrbrecords_fetch_lagrecordrrlendrain) rirpartrrrr part_recordsrs r:rFetcher._appendds  ! !""..r22 II89; =$$11"55 II<=? A **55b9BBH%%8 *+3??B@#yy5  K&&|4 !//::2>HH (T]]MM33::9G]G];]^!IIi "8"8$:K:KMBSTXTjTjlnprBsD''2226?<((  89;=S=S"//+ r9cURRU5(d[RSU5 gURR U5(d[RSU5 gU(a=X RR UR :Xd[RSU5 g[RSX5 URRX5 g)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.) rZrrsrtis_offset_reset_neededrrinfoseek)rirrrs r:_reset_offset_if_neededFetcher._reset_offset_if_neededs""..y99 IIUW` a$$;;IFF IIVXa b y,?,?,J,J9,U,d,dd IIdfo p HHF Z    $ $Y 7r9c^TRU5n[R"U5Hup4TRR U5(dM'[ UR 55n[R"5TRSS- -nTRRXV5 U4SjnU4SjnTRX45n U RXt5 U RX5 M g)NrNrcj>Uup#U(afTRRU[R"5TRSS- -5 TRR R 5 [R"U5H'upEXupgTRXFUR5 M) g)NrOr) rZ reset_failedryrUrYrrrrqrr) timestamps_and_epochsresultrpartitions_to_retryrrr_epochris r: on_success0Fetcher._reset_offsets_async..on_successs7=4&''445H$))+X\XcXcdvXwz~X~J~LL((779),)G%I!6!AJB00 N*Hr9cV>TRRU[R"5TRSS- -5 TRR R 5 [USS5(d0TR(dUTlg[RSU5 gg)NrOrrFzKDiscarding error in ListOffsetResponse because another error is pending: %s) rZrryrUrYrrgetattrrgrserror)rrris r: on_failure0Fetcher._reset_offsets_async..on_failures##00TYY[4;;WiKjmqKq=qr $$335uk599>>>C; "oqvw :r9)_group_list_offset_requestsrrqrYreadyrekeysryrUrZset_reset_pending_send_list_offsets_requestrwrz) rirtimestamps_by_noderrr expire_atrrrs ` r:rFetcher._reset_offsets_asyncs!==jI.1mmUS==S-ss'TRUS5 TRUS5 US(d&TR(dTRTT45 gggNrr)rrsuccess)remaining_responsesr!rlist_offsets_futurers r:r7Fetcher._send_list_offsets_requests..on_successse  "a ' "  " "58 ,  & &uQx 0&q)2E2M2M#++_>Q,RS3N)r9cL>TR(dTRU5 ggr)rfailure)errrs r:on_fail4Fetcher._send_list_offsets_requests..on_fails&..#++C0/r9) rrr rW StaleMetadatarrerrrqrrwrz) rirrrrr r_frrrs @@@r:r#Fetcher._send_list_offsets_requestss"==jI!8##F$8$8$:; ;%h&!e"#567 T 1$'==1C#D G00EB OOJ < NN7 #$E#"r9c4[R"[5n[R"U5Hup4UR R RU5nUcaUR RUR5 [RSU5 UR R R5 MUS:Xa<[RSU5 UR R R5 MSnXF4X%U'M [U5$)Nz+Partition %s is unknown for fetching offsetrzRLeader for partition %s unavailable for fetching offset, wait for metadata refresh) r[rrrrqrYrleader_for_partition add_topicrrsrtr)rirrrrrrs r:r#Fetcher._group_list_offset_requestss(44T:$'MM*$= Ill**?? JG &&y7 GS $$335B >?HJ $$335! :C9R"+I6%>&''r9c v^ URR[SS9nURSS:XaUS:a[R "S5e[ R"[5n[R"U5H^unupgUS:aURXv4nO$US:aURU4nOURUS4nXERRU5 M` US::a.[U"S[[R"U555n O8[U"SUR[[R"U555n [5m [ R#S X5 URR%X5n U R'UR(T 5 U R+U 4S j5 T $) N max_versionrQrz@read_committed isolation level requires ListOffsetsRequest >= v2rrz)Sending ListOffsetRequest %s to broker %sc&>TRU5$r)r )rrs r:4Fetcher._send_list_offsets_request..s!2r9)rY api_versionr rUrWUnsupportedVersionErrorr[rrrrqrrr~rcrrsrtrvrw_handle_list_offsets_responserz) rirrversionby_topicrrrdatarrrs @r:r"Fetcher._send_list_offsets_requestsp,,**+=1*M ;;( )-= ='A+001st t**40-0]];P-Q )B)!| l>A i0 i3 XX  % %d +.R a<(1x013G)1))x013G =wP \\  w 0 ::FC 23 r9c [5n[5n[5nURGH upgUGHnUSSup[Xi5n [R "U 5n U [R LaURS:Xa3USn [U 5S::dS5eU (d[n OU Sn SnSnO"URS::a USSupSnOUSSupn[RSXX5 U [:wa[XU5X9'MMU [RLa[RS U 5 GMU [R[R[R 4;a5[RS U R"U 5 UR%U 5 GMkU [R&La*[R)S U 5 UR%U 5 GMU [R*LaUR%U5 GM[R)S XR"5 UR%U 5 GM GM U(a&UR-[R*"U55 gUR/X445 g) zCallback for the response of the ListOffsets api call Arguments: future (Future): the future to update based on response response (ListOffsetsResponse): response from the server Raises: AssertionError: if response does not match partition Nrrrz,Expected ListOffsetsResponse with one offsetrz^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)rretopicsrrWfor_codeNoError API_VERSIONrrrsrtr UnsupportedForMessageFormatErrorNotLeaderForPartitionErrorReplicaNotAvailableErrorKafkaStorageErrorr4ruUnknownTopicOrPartitionErrorwarningTopicAuthorizationFailedErrorr r)rirresponserrunauthorized_topicsr part_datapartition_infor error_code error_typerrrrs r:r%Fetcher._handle_list_offsets_response!s?&!e!e ( E"+(6r(:% *5< #__Z8 /++q0"0"3"7|q0`2``0&%3F%,QZF$( ') !--2,:12,>) ') :H:L7 <IIQ'J/5G[g5h206#J#JJIIIJSUF$E$E$*$C$C$*$<$<$>>II23=3F3F S'++I66#F#FFKK!)+45(++I66#G#GG'++E2KK!&'02E2EG'++I6_#,!0b  NN6??@ST U NNOA Br9cLURR5n[R"UR5Vs1sHo"RiM nnUR nU(aUR UR5 UVs/sH oUU;dM UPM sn$s snfs snfr)rZfetchable_partitionsrSr]r(r^ru)ri fetchablefetchdiscardcurrentrs r:_fetchable_partitionsFetcher._fetchable_partitionsds''<<> 7;ii@W@W6XY6XU((6XY..  KK// 0&5n US::a8[U"SUR,SUR,SU R@5n GOUS:XaE[U"SUR,SUR,SUR,SU R@5n OUS::aP[U"SUR,SUR,SUR,SURBU R@5n Op[U"SUR,SUR,SUR,SURBU RDU RFU R@U RH5 n 0n [0R2"U5HupUS ::aU SnOU SnXU 'M X4Xt'GM U$)aCreate 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}), ...} (version depends on client api_versions) rNrz5 UR@RBRE5 GOU[RFLaUR RURnUbURU:wa#[RSX#UR5 GO1UR RI5(a2[RKSX25 UR RMU5 O[RF"X#05eU[RNLaJ[RQSURR5 [RN"[UURR/55e[WUSS5(aR[RSX&"55 [WUSS5(a$UR@RBRE5 OU"S5eUc.UR(aURRUSS5 U[RLaUR RU5 U$!Uc.UR(aURRUSS5 U[RLaUR RU5 ff=f) NrzIIgnoring 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 %drr rzBPreparing to read %s bytes of data for partition %s with offset %drDrErK)rDrErKrQaborted_transactionsr,on_drainr$z/There are some messages at [Partition=Offset]: z+ whose size is larger than the fetch size rIz 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)z,Failed to make progress reading messages at =zZ. Received a non-empty fetch response from the server, but no complete records were found.zError fetching partition %s: %szqDiscarding stale fetch response for partition %s since the fetched offset %d does not match the current offset %dz6Fetch offset %s is out of range for topic-partition %sz%Not authorized to read from topic %s.rFz)Retriable error fetching partition %s: %srz$Unexpected error while fetching data),r(r)r+rWr&rZrrsrtr'rrrr,rmove_partition_to_endrr* size_in_bytesPartitionRecordsrUrc_on_partition_records_drainhas_nextr= KafkaErrorrr*r+r-r,r4rYrrOffsetOutOfRangeErrorhas_default_offset_reset_policyrrequest_offset_resetr/r.rrer) rir\rr[r4rr5parsed_recordsrrecordsrprecord_too_large_partitionss r:rFetcher._parse_fetched_datas  , ,&55 / > >r B __Z0 ] >&&33B77 =>@Bv~~-  ..99"=FF#x,'FII457&oo/ J%/*K*K1188QB/##99"=0M((F(Fr(JK'+$"33r9+:+I+I"+M($55:+:+I+I"+M( ^!//12E!%!6!6|HL TfHgJN++VjJkBF++lB[GKG\G\L`IXIjIj@D@`@`"7"b''))g.C.C.E.I&77!;796H31 !< $ ,G H J899%//SUWc1eff>CL''2226@ A A & ? ? & C C & 8 8 :: ;RATATU $$335v;;;..99"=FF#x,'FII346hooW((HHJJHHUWch''<sr9c XlX lSUlXlSUlSUlXpl[5Ul[R"[U(aUV s/sH n [U 6PM sn O/SS95Ul XlX`l[ R""UR$UR'X#XE55UlXlSUlgs sn f)NrrcUR$r) first_offset)txns r:r3Fetcher.PartitionRecords.__init__..s s'7'7r9)r )r[r(rrr records_readrQreaborted_producer_idsr[r\sortedr rpr,rK itertools dropwhile_maybe_skip_record_unpack_recordsrecord_iteratorrq_next_inline_exception) rir[rr}rDrErKrQrpr,rqr!s r:rm!Fetcher.PartitionRecords.__init__s !- #% "D %1 "DO !D #2 (+D %(3(9(9Wk>RS>Rd*D1>RSqs79)D %&7 "(O#,#6#6''$$R2BW$YD %M*.D 'TsC cURUR:a,[RSURUR5 gg)Nz*Skipping message offset: %s (expecting %s)TF)rr[rsrt)rirs r:r+Fetcher.PartitionRecords._maybe_skip_records: }}t000 F --):):<r9cURSL$r)rrs r:__bool__!Fetcher.PartitionRecords.__bool__s''t3 3r9cURblSUlSUlUR(a;URRURUR UR 5 URU5 ggr)rrr,rr(rrrqrs r:rFetcher.PartitionRecords.drainsa##/'+$.2+))**11$2F2FY]YjYjk d# 0r9cPUR(aURSsolUegr)r)rirs r:"_maybe_raise_next_inline_exception;Fetcher.PartitionRecords._maybe_raise_next_inline_exceptions)**373N3NPT00 +r9cUR5 /nUR[R"URSU55 U$![ anU(dUeX0lSnAU$SnAff=fr)rrrislicerrr)rinr}rs r:rFetcher.PartitionRecords.takesh  3 3 5G 0y//0D0DaKL N  0G./++N  0s1A A+A&&A+c#$# UR5nSnUGb~UnUR(aHUR5(d3[R"SUR <SUR <S35eURS:XGa9URUlUR[:XaUR5(aURUR5 URnURU5(aUR R#U5 OpUR'U5(aZ[(R+SUR XuR UR5 UR,UlUR5nGMtUR0(a$UR,UlUR5nGMUGHnUR(aHUR5(d3[R"SUR <SUR2<S35eUR4b[7UR45OSn UR8b[7UR85OSn UR;X1R<UR45n UR;XAR<UR85n UR>n U (a[ASU 55OSnU=RBS - sl!U=RDURF- sl"UR2S -Ul[IUR<URJURUR2URLURNXXRPXU5 v GM UR5nUbGM~U(a!URS:XaUR,UlURS5 g![$a GNTf=f![Ta! [(RWS 5 [YS 5ef=f7f) NzRecord batch for partition z at offset z failed crc checkrzXSkipping aborted record batch from partition %s with producer_id %s and offsets %s to %szRecord for partition rc3|# UH2up[URS55Ub [U5OS-v M4 g7f)zutf-8Nr)rencode).0h_keyh_vals r: ;Fetcher.PartitionRecords._unpack_records.. s;*%#gsfkC W 56HY#e*_`a#s:<rz)StopIteration raised unpacking messageset)- next_batchrK validate_crcrWCorruptRecordErrorr( base_offsetmagicrrQREAD_COMMITTEDhas_producer_id#_consume_aborted_transactions_up_to last_offset producer_id_contains_abort_markerrrhri_is_batch_abortedrsrt next_offsetris_control_batchrr rr! _deserializerr"sumrrrtrrrrr#r StopIterationr. RuntimeError)rirr}rDrEbatch last_batchrrkey_size value_sizer r!r" header_sizes r:r(Fetcher.PartitionRecords._unpack_recordss^P P**,! '!&Ju/A/A/C/C$77$($8$8%:K:K!MNN{{a',1,>,>)//>AeF[F[F]F]!DDUEVEVW*/*;*;K#::5AA!)$($=$=$D$D[$Q"&!7!7!>!> # +>*.*>*> M^M^`e`q`q!s:?9J9J 6(/(:(:(< (!115:5F5FD2$+$6$6$8E$"'??63F3F3H3H"(";";(,(<(OI [RSURURXGX`R R55 [R"UVs0sHoUXG-;dM XQU_M sn5n[XUR5$s snf)zm Arguments: next_partitions (dict): TopicPartition -> 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 %s)ris_fullrsrtrrrrKrerpoprur[rE) rirSprev_tpsnext_tpsaddedrremovedalteredrNs r:rJFetchSessionHandler.build_nextVs    % % IIM""DLL#o2F H&5 ##OT4;M;MN Nt..3356++-. OQYd#B*9*=D # #B '%B  # # ' ' +%%B"d&=&=b&AA.=.A''+ B& f$$dllEGMdMdMiMiMk m))_*x_rglgv`w+B2r/B+B_*xy$2D2DEE+ys - G"< G"c UR[RR:wa[R"UR5n[ R SURURU"55 U[RLa[RUlgURR5UlgURU5n[URR!55nURR"(GaBX4:wa<[ R SURX4- XC- 5 [RUlgUR$[R&:Xa@[ R)SUR[+U55 [RUlgUR$[R,:Xa+[ R)SUR[+U55 g[ R)SURUR$[+U55 [R/UR$5UlgX4- (aC[ R SURX4- 5 URR5UlgUR$[R&:Xau[ R)S URURR$[+U5[+UR5[+U5- 5 [RUlgUR$[R,:Xa+[ R)S UR[+U55 g[ R)S URUR$[+U5[+UR5[+U5- 5 URR15Ulg) Nz?? " == (> !r9cNURURUR5$r) __class__rrrs r:r!FetchMetadata.next_close_existings~~doot/A/ABBr9cDU"XRUR55$r)r r)rrs r:rFetchMetadata.new_incrementals:~~c.?.?@AAr9clURURURUR55$r)r rr rPrs r:rFetchMetadata.next_incrementals$~~doottzz/JKKr9)rPrN)r4r5r6r7 __slots__rrrrrrmpropertyr classmethodr rrrr8r2r9r:rLrLss'IIMKRR""CBBLr9rLcn\rSrSrSrSr\S5r\S5r\S5r \S5r \S5r S r g ) rKi)_to_send _to_forget _metadatacpU=(d [5UlU=(d [5UlX0lgr)rrrerr)rirNrQmetadatas r:rmFetchRequestData.__init__s#)46 #,su!r9cUR$r)rrs r:rFetchRequestData.metadatas ~~r9c.URR$r)rrrs r:rOFetchRequestData.ids~~(((r9c.URR$r)rrPrs r:rPFetchRequestData.epochs~~###r9c[R"[5n[R"UR 5H"up#XR RU5 M$ [UR55$r) r[rrrrqrrr~items)rir+rr3s r:rNFetchRequestData.to_sendsW%006"%-- "> B 88 $ + +N ;#?N((*++r9c[R"[5nURH*nXRR UR 5 M, [UR55$r)r[rrrrr~rr$)rir+rs r:rQFetchRequestData.to_forgetsO%006//B 88 $ + +BLL 9"N((*++r9)rrrN) r4r5r6r7rrmrrrOrPrNrQr8r2r9r:rKrKsp7I" ))$$,,,,r9rKc\rSrSrSrSrSrg) FetchMetricsi total_bytes total_recordsc SUlSUlgrr*rs r:rmFetchMetrics.__init__sr9N)r4r5r6r7rrmr8r2r9r:r)r)s 0Ir9r)c$\rSrSrSrSrSrSrg)rXiz 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. cxXlX l[5Ul[R "[5Ulgr)sensorsunrecorded_partitionsr) fetch_metricsr[rtopic_fetch_metrics)rir1rs r:rm&FetchResponseMetricAggregator.__init__ s* %/")^#.#:#:<#H r9c URRU5 UR=RU- slUR=RU- slUR UR =RU- slUR UR =RU- slUR(dURRRURR5 URRRURR5 [R"UR 5H5upEURRXERUR5 M7 gg)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)r2rhr3r+r,r4rr1 bytes_fetchedrrecords_fetchedrrqrecord_topic_fetch_metrics)rir num_bytes num_recordsrrLs r:r$FetchResponseMetricAggregator.record&s "")))4 &&)3& ((K7(   1==J=   1??;N?)) LL & & - -d.@.@.L.L M LL ( ( / /0B0B0P0P Q"%--0H0H"I 77?R?RT[TiTij#J*r9)r3r1r4r2N)r4r5r6r7rrmrr8r2r9r:rXrXs I kr9rXc \rSrSrSrSrSrg)rai:cfXlU<S3UlURS5UlURR UR SURS5[ 55 URR UR SURS5[55 URR UR SURS5[55 URRS 5Ul URR UR S URS 5[ 55 URR UR S URS 5[55 URS5Ul URR UR SURS5[ 55 URR UR SURS5[55 URR UR SURS5[[5S95 URS5Ul URR UR SURS5[55 g)Nz-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) rL group_namesensorr7ru metric_namerr r r8rYrr)rirLprefixs r:rmFetchManagerMetrics.__init__;s7 8>@$^^O< w223CT__ = ?@C G w223CT__ = ?@C G w223H$// = ?@D H $||223DE   !4!45NPTP_P_ ;"=>Ae E   !4!45Ldoo ?"ABF& J%^^O< w223F 9 ; ""7#6#67H$// \$^_b_d fr9cSRSUS/5nURRU5nU(dSURSS50nURR U5nUR URR SURSU<3U5[55 UR URR SURSU<3U5[55 UR URR S URS U<3U5[55 URU5 SRSUS /5nURRU5nU(dSURSS50nURR U5nUR URR S URS U<3U5[55 UR URR SURSU<3U5[55 URU5 g)N.rr?rjr@z:The average number of bytes fetched per request for topic rAz:The maximum number of bytes fetched per request for topic rBz:The average number of bytes consumed per second for topic rCrDz8The average number of records in each request for topic rEzrcs0  55A')OO !)$ ''(8vw ''(8,-  **+>02  1 1  &++ F cllF R |2&|2~"LF"LH&m&F&F HcHcd $]%E%E}G`G`a $,v$,N6kFkBD,&D,r9