package Ġcom . info bip . k af k ist ry . kafka . records ampling Ċ Ċ import Ġorg . apache . kafka . clients . consumer . Kafka Consumer Ċ import Ġorg . apache . kafka . common . Topic Partition Ċ import Ġcom . info bip . k af k ist ry . utils . do With No Logging Overrides Ċ import Ġcom . info bip . k af k ist ry . utils . do With Only Error Logging Ċ import Ġcom . info bip . k af k ist ry . kafka . Partition Ċ import Ġcom . info bip . k af k ist ry . kafka . Partition Offsets Ċ import Ġcom . info bip . k af k ist ry . kafka . Record Visitor Ċ import Ġcom . info bip . k af k ist ry . kafka . Sampling Position Ċ import Ġcom . info bip . k af k ist ry . model . Topic Name Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġjava . time . Duration Ċ Ċ /** Ċ Ġ* ĠRecord Ġsap ling Ġworks Ġby Ġassigning Ġconsumer Ġto Ġevery Ġnon - empty Ġtopic Ġpartition Ġfrom Ġwhich Ġsample Ġis Ġwanted Ġto Ġbe Ġtaken . Ċ Ġ* ĠThen Ġpooling Ġis Ġdone Ġin Ġloop Ġuntil Ġone Ġrecord Ġfrom Ġevery Ġpartition Ġis Ġreceived . Ċ Ġ* ĠAfter Ġrecord Ġis Ġreceived Ġfrom Ġone Ġpartition , Ġconsumer Ġis Ġun subscribed Ġfrom Ġthat Ġpartition Ġso Ġthat Ġfurther Ġpool Ġcalls Ċ Ġ* Ġtarget Ġonly Ġremaining Ġpartitions . Ċ Ġ*/ Ċ class ĠRecord Read Sampler ( ĊĠĠĠ Ġprivate Ġval Ġconsumer : ĠKafka Consumer < ByteArray , ĠByteArray >, ĊĠĠĠ Ġinitial Pool Timeout : ĠLong , ĊĠĠĠ Ġpool Timeout : ĠLong , Ċ ) Ġ: ĠAuto Closeable Ġ{ ĊĊĠĠĠ Ġprivate Ġval Ġlog Ġ= ĠLoggerFactory . getLogger ( Record Read Sampler :: class . java ) ĊĊĠĠĠ Ġprivate Ġval Ġinitial Timeout Ġ= ĠDuration . of Millis ( initial Pool Timeout ) ĊĠĠĠ Ġprivate Ġval Ġpool Timeout Ġ= ĠDuration . of Millis ( pool Timeout ) ĊĊĠĠĠ Ġoverride Ġfun Ġclose () Ġ= Ġconsumer . close ( initial Timeout ) ĊĊĠĠĠ Ġfun Ġtest () Ġ= ĠUnit ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠPerform Ġround Ġof Ġsampling Ġfor Ġevery Ġgiven Ġ[ topic Partition Offsets ]. ĊĠĠĠĠ Ġ* ĠS ampling Ġwill Ġsample Ġat - most Ġ` 1 ` Ġrecords Ġfrom Ġevery Ġtopic - partition . ĊĠĠĠĠ Ġ* ĠIt Ġwill Ġbe Ġeither Ġone ĠO L DEST Ġor Ġone ĠNEW EST Ġ( but Ġalready Ġproduced ) Ġrecord Ġin Ġpartition , ĊĠĠĠĠ Ġ* Ġdepending Ġon Ġ[ sampling Position ] Ġargument . ĊĠĠĠĠ Ġ* ĠResult Ġis Ġgiven Ġvia Ġ[ record Visitor ] Ġcallback Ġto Ġallow Ġprocessing Ġof Ġresults Ġas Ġsoon Ġas Ġavailable Ġto ĊĠĠĠĠ Ġ* Ġavoid Ġneed Ġto Ġcollect Ġeverything Ġin Ġmemory Ġat Ġonce . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġread Sample Records ( ĊĠĠĠĠĠĠĠ Ġtopic Partition Offsets : ĠMap < Topic Name , ĠMap < Partition , ĠPartition Offsets >>, ĊĠĠĠĠĠĠĠ Ġsampling Position : ĠS ampling Position , ĊĠĠĠĠĠĠĠ Ġrecord Visitor : ĠRecord Visitor , ĊĠĠĠ Ġ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġall Needed Partitions Offsets Ġ= Ġextract Needed Partitions ( topic Partition Offsets ) ĊĠĠĠĠĠĠĠ Ġlog . info (" Go ing Ġto Ġsample Ġrecords Ġfrom Ġ{} Ġtopic Ġpartitions , Ġposition : Ġ{}", ĊĠĠĠĠĠĠĠĠĠĠĠ Ġall Needed Partitions Offsets . size , Ġsampling Position ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġvar Ġcount Ġ= Ġ 0 ĊĠĠĠĠĠĠĠ Ġdo With Only Error Logging Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdo Sample Records ( all Needed Partitions Offsets , Ġsampling Position , Ġrecord Visitor . compose Ġ{ Ġcount ++ Ġ}) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġlog . info (" Sample d Ġ{} Ġrecords Ġfrom Ġ{} Ġtopic Ġpartitions , Ġposition : Ġ{}", ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcount , Ġall Needed Partitions Offsets . size , Ġsampling Position ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġdo Sample Records ( ĊĠĠĠĠĠĠĠ Ġall Needed Partitions Offsets : ĠMap < Topic Partition , ĠPartition Offsets >, ĊĠĠĠĠĠĠĠ Ġsampling Position : ĠS ampling Position , ĊĠĠĠĠĠĠĠ Ġrecord Visitor : ĠRecord Visitor , ĊĠĠĠ Ġ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġremaining Partitions Ġ= Ġall Needed Partitions Offsets . keys . to Mutable Set () ĊĠĠĠĠĠĠĠ Ġinitialize Consumer ( all Needed Partitions Offsets , Ġsampling Position ) ĊĠĠĠĠĠĠĠ Ġvar Ġre wind Attempt Ġ= Ġ 0 ĊĠĠĠĠĠĠĠ Ġvar Ġtimeout Ġ= Ġinitial Timeout ĊĠĠĠĠĠĠĠ Ġvar Ġpool Attempts Left Ġ= Ġall Needed Partitions Offsets . size ĊĠĠĠĠĠĠĠ Ġwhile Ġ( pool Attempts Left Ġ> Ġ 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpool Attempts Left -- ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġrecord Ġ= Ġconsumer . poll ( timeout ). first OrNull () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( record Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtimeout Ġ= Ġpool Timeout ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrecord . headers (). hashCode () Ġ// force Ġeager Ġinitialization Ġof Ġthread Ġunsafe ĠRecord Header Ġbyte Buffer ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdo With No Logging Overrides Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrecord Visitor . visit ( record ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtopic Partition Ġ= ĠTopic Partition ( record . topic (), Ġrecord . partition ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġgot New Ġ= Ġremaining Partitions . remove ( topic Partition ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( remaining Partitions . isEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbreak ĠĠ Ġ// we Ġcollected Ġrecords Ġfrom Ġall Ġneeded Ġpartitions ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( got New ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer . assign ( remaining Partitions ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( sampling Position ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠS ampling Position . OL DEST Ġ-> Ġbreak ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠS ampling Position . NEW EST Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( rewind Attempt Ġ>= Ġ 2 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbreak ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpool Attempts Left Ġ+= Ġremaining Partitions . size ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġre wind Attempt ++ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġre wind Consumer ( all Needed Partitions Offsets , Ġremaining Partitions , Ġre wind Attempt ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġinitialize Consumer ( ĊĠĠĠĠĠĠĠ Ġall Needed Partitions Offsets : ĠMap < Topic Partition , ĠPartition Offsets >, ĊĠĠĠĠĠĠĠ Ġsampling Position : ĠS ampling Position ĊĠĠĠ Ġ) Ġ{ ĊĠĠĠĠĠĠĠ Ġconsumer . assign ( all Needed Partitions Offsets . keys . toSet ()) ĊĠĠĠĠĠĠĠ Ġall Needed Partitions Offsets . forEach Ġ{ Ġ( topic Partition , Ġoffsets ) Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġoffset Ġ= Ġwhen Ġ( sampling Position ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠS ampling Position . OL DEST Ġ-> Ġoffsets . begin ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠS ampling Position . NEW EST Ġ-> Ġoffsets . end . minus ( 1 ). coerce At Least ( offsets . begin ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer . seek ( topic Partition , Ġoffset ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠNeed Ġto Ġre wind Ġbackwards Ġbecause Ġsome Ġtopic Ġpartitions Ġmight Ġhave Ġfew Ġnewest Ġoffsets Ġbeing Ġ" used " Ġfor ĊĠĠĠĠ Ġ* Ġtransaction Ġmarker Ġand Ġtherefore Ġreading Ġfrom Ġ` end - 1 ` Ġwon 't Ġyield Ġwith Ġany Ġrecord Ġ( until Ġnew Ġone Ġis Ġproduced ). ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġprivate Ġfun Ġre wind Consumer ( ĊĠĠĠĠĠĠĠ Ġall Needed Partitions Offsets : ĠMap < Topic Partition , ĠPartition Offsets >, ĊĠĠĠĠĠĠĠ Ġremaining Partitions : ĠMutable Set < Topic Partition >, ĊĠĠĠĠĠĠĠ Ġre wind Attempt : ĠInt ĊĠĠĠ Ġ) Ġ{ ĊĠĠĠĠĠĠĠ Ġall Needed Partitions Offsets ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Keys Ġ{ Ġit Ġin Ġremaining Partitions Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. forEach Ġ{ Ġ( topic Partition , Ġoffsets ) Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġnew Offset Ġ= Ġoffsets . end . minus ( rewind Attempt Ġ* Ġ 2 ). coerce At Least ( offsets . begin ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer . seek ( topic Partition , Ġnew Offset ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġextract Needed Partitions ( ĊĠĠĠĠĠĠĠ Ġtopic Partition Offsets : ĠMap < Topic Name , ĠMap < Partition , ĠPartition Offsets >> ĊĠĠĠ Ġ): ĠMap < Topic Partition , ĠPartition Offsets > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġtopic Partition Offsets ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ Ġ( topic , Ġpartitions Offsets ) Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpartitions Offsets ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Values Ġ{ Ġit . end Ġ> Ġit . begin Ġ} Ġ// only Ġpartitions Ġwhich Ġhave Ġrecords ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġ( partition , Ġoffsets ) Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠTopic Partition ( topic , Ġpartition ) Ġto Ġoffsets ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. associate Ġ{ Ġit Ġ} ĊĠĠĠ Ġ} Ċ Ċ } Ċ