package Ġkafka . flow . consumer . with . group . id Ċ Ċ import Ġbe . delta . flow . time . milliseconds Ċ import Ġbe . delta . flow . time . seconds Ċ import Ġjava . time . Instant Ċ import Ġjava . util . Properties Ċ import Ġjava . util . concurrent . Concurrent HashMap Ċ import Ġkafka . flow . consumer . Auto Stop Policy Ċ import Ġkafka . flow . consumer . EndOf Batch Ċ import Ġkafka . flow . consumer . Kafka Flow Consumer With GroupId Ċ import Ġkafka . flow . consumer . Kafka Message Ċ import Ġkafka . flow . consumer . Partition Changed Message Ċ import Ġkafka . flow . consumer . Partitions Assigned Ċ import Ġkafka . flow . consumer . Partitions Rev oked Ċ import Ġkafka . flow . consumer . Record Ċ import Ġkafka . flow . consumer . Start Consum ing Ċ import Ġkafka . flow . consumer . Start Offset Policy Ċ import Ġkafka . flow . consumer . Stop Consum ing Ċ import Ġkotlinx . coroutines . Cancellation Exception Ċ import Ġkotlinx . coroutines . Coroutine Scope Ċ import Ġkotlinx . coroutines . Dispatch ers Ċ import Ġkotlinx . coroutines . Job Ċ import Ġkotlinx . coroutines . Non C ancellable Ċ import Ġkotlinx . coroutines . channels . Channel Ċ import Ġkotlinx . coroutines . current Coroutine Context Ċ import Ġkotlinx . coroutines . delay Ċ import Ġkotlinx . coroutines . flow . Flow Ċ import Ġkotlinx . coroutines . flow . consume As Flow Ċ import Ġkotlinx . coroutines . flow . on Completion Ċ import Ġkotlinx . coroutines . isActive Ċ import Ġkotlinx . coroutines . launch Ċ import Ġkotlinx . coroutines . sync . Mutex Ċ import Ġkotlinx . coroutines . sync . with Lock Ċ import Ġkotlinx . coroutines . with Context Ċ import Ġkotlinx . coroutines . yield Ċ import Ġorg . apache . kafka . clients . consumer . Consumer Config Ċ import Ġorg . apache . kafka . clients . consumer . Consumer Re balance Listener Ċ import Ġorg . apache . kafka . clients . consumer . Kafka Consumer Ċ import Ġorg . apache . kafka . clients . consumer . Offset And Metadata Ċ import Ġorg . apache . kafka . common . Topic Partition Ċ import Ġorg . apache . kafka . common . serialization . ByteArray Deserializer Ċ import Ġorg . slf 4 j . LoggerFactory Ċ Ċ public Ġclass ĠKafka Flow Consumer With GroupId Impl ( ĊĠĠĠ Ġclient Properties : ĠProperties , ĊĠĠĠ Ġprivate Ġval Ġtopics : ĠList < String >, ĊĠĠĠ Ġprivate Ġval Ġstart Offset Policy : ĠStart Offset Policy , ĊĠĠĠ Ġprivate Ġval Ġauto Stop Policy : ĠAuto Stop Policy Ċ ) Ġ: ĠKafka Flow Consumer With GroupId < Kafka Message < Unit , ĠUnit , ĠUnit , ĠUnit , ĠWithout Transaction >> Ġ{ ĊĠĠĠ Ġprivate Ġval Ġproperties Ġ= Ġproperties ( client Properties ) ĊĠĠĠ Ġprivate Ġval Ġlogger Ġ= ĠLoggerFactory . getLogger ( Transaction Manager :: class . java ) ĊĠĠĠ Ġprivate Ġval Ġdelegate : ĠKafka Consumer < ByteArray , ĠByteArray > Ġ= ĠKafka Consumer ( properties , ĠByteArray Deserializer (), ĠByteArray Deserializer ()) ĊĠĠĠ Ġprivate Ġvar Ġrunning Ġ= Ġfalse ĊĠĠĠ Ġprivate Ġvar Ġstop Requested : ĠBoolean Ġ= Ġfalse ĊĠĠĠ Ġprivate Ġvar Ġstart Instant : ĠInstant ? Ġ= Ġnull ĊĠĠĠ Ġprivate Ġvar Ġend Offsets : ĠMap < Topic Partition , ĠLong > Ġ= Ġempty Map () ĊĠĠĠ Ġprivate Ġvar Ġ_ assignment : ĠList < Topic Partition > Ġ= Ġempty List () ĊĠĠĠ Ġprivate Ġval Ġpartition Changed Messages Ġ= Ġmutable ListOf < Partition Changed Message < Unit , ĠUnit , ĠUnit , ĠUnit , ĠWithout Transaction >>() ĊĠĠĠ Ġprivate Ġval Ġdelegate Mutex Ġ= ĠMutex () ĊĠĠĠ Ġprivate Ġval Ġpoll Duration Ġ= Ġ 1 0 0 . milliseconds () ĊĠĠĠ Ġprivate Ġval Ġpositions Ġ= ĠConcurrent HashMap < Topic Partition , ĠLong >() ĊĠĠĠ Ġoverride Ġval Ġassignment : ĠList < Topic Partition > Ġget () Ġ= Ġ_ assignment ĊĠĠĠ Ġprivate Ġvar Ġend Offsets Loop : ĠJob ? Ġ= Ġnull ĊĊĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠ Ġrequire NotNull ( client Properties [ Consumer Config . GROUP _ ID _ CONFIG ]) Ġ{ Ġ"${ Consumer Config . GROUP _ ID _ CONFIG } Ġmust Ġbe Ġset " Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġsuspend Ġfun Ġstart Consum ing ( on Deserialization Exception : Ġsuspend Ġ( Throwable ) Ġ-> ĠUnit ): ĠFlow < Kafka Message < Unit , ĠUnit , ĠUnit , ĠUnit , ĠWithout Transaction >> Ġ{ ĊĠĠĠĠĠĠĠ Ġsubscribe () ĊĠĠĠĠĠĠĠ Ġreturn Ġcreate Consumer Channel () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. consume As Flow () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. on Completion Ġ{ Ġcleanup () Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġis Running (): ĠBoolean Ġ= Ġrunning ĊĊĠĠĠ Ġoverride Ġfun Ġis Up ToDate (): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( stop Requested ) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠ Ġif Ġ(! running ) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠ Ġval Ġlag Ġ= Ġlag () ĊĠĠĠĠĠĠĠ Ġif Ġ( lag Ġ== Ġnull Ġ|| Ġlag Ġ> Ġ 0 ) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠ Ġreturn Ġtrue ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġl ags (): ĠMap < Topic Partition , ĠLong ?> ? Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ(! isRunning ()) Ġreturn Ġnull ĊĠĠĠĠĠĠĠ Ġif Ġ(_ assignment . isEmpty ()) Ġreturn Ġempty Map () ĊĠĠĠĠĠĠĠ Ġreturn Ġ_ assignment . associate With Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġend Offset Ġ= Ġend Offsets [ it ] ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġposition Ġ= Ġpositions [ it ] ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( position Ġ!= Ġnull Ġ&& Ġend Offset Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ( end Offset Ġ- Ġposition ). coerce At Least ( 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġif Ġ( end Offset Ġ== Ġ 0 L ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ 0 L ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġlag (): ĠLong ? Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġl ags : ĠMap < Topic Partition , ĠLong ?> ? Ġ= Ġl ags () ĊĠĠĠĠĠĠĠ Ġif Ġ( l ags Ġ== Ġnull Ġ|| Ġl ags . values . contains ( null )) Ġreturn Ġnull ĊĠĠĠĠĠĠĠ Ġreturn Ġl ags . values . filter NotNull (). sum () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġstop () Ġ{ ĊĠĠĠĠĠĠĠ Ġstop Requested Ġ= Ġtrue ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġclose () Ġ{ ĊĠĠĠĠĠĠĠ Ġstop () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġsuspend Ġfun Ġcommit ( offsets To Commit : ĠMap < Topic Partition , ĠOffset And Metadata >, Ġcallback : Ġ( Result < Map < Topic Partition , ĠOffset And Metadata >>) Ġ-> ĠUnit ) Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( offsets To Commit . isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdelegate Mutex . with Lock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelegate . commit Async ( offsets To Commit ) Ġ{ Ġoffsets , Ġexception Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( exception Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcallback . invoke ( Result . failure ( exception )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcallback ( Result . success ( offsets !! )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġsuspend Ġfun Ġrollback ( topic Partition To Rollback : ĠSet < Topic Partition >) Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( topic Partition To Rollback . isEmpty ()) Ġreturn ĊĠĠĠĠĠĠĠ Ġdelegate Mutex . with Lock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcheck ( isRunning ()) Ġ{ Ġ" Cannot Ġrollback Ġtransaction Ġwhen Ġthe Ġconsumer Ġisn 't Ġrunning " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcommitted Offsets Ġ= Ġdelegate . committed ( topic Partition To Rollback ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtopic Partition To Rollback . forEach Ġ{ Ġtopic Partition Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelegate . seek ( topic Partition , Ġcommitted Offsets [ topic Partition ]?. offset () Ġ?: Ġ 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġsuspend Ġfun Ġcreate Consumer Channel (): ĠChannel < Kafka Message < Unit , ĠUnit , ĠUnit , ĠUnit , ĠWithout Transaction >> Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġchannel Ġ= ĠChannel < Kafka Message < Unit , ĠUnit , ĠUnit , ĠUnit , ĠWithout Transaction >>() ĊĠĠĠĠĠĠĠ ĠCor outine Scope ( current Coroutine Context ()). launch ( Dispatch ers . IO ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġchannel . send ( Start Consum ing ( this @ Kafka Flow Consumer With GroupId Impl )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdo Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfetch And Process Records ( channel ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġwhile Ġ(! should Stop () Ġ&& ĠisActive ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( isActive ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġchannel . send ( Stop Consum ing ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( t : ĠC ancellation Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( t : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġt . printStackTrace () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġchannel . close () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġend Offsets Loop ?. cancel () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġchannel ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġsuspend Ġfun Ġfetch And Process Records ( channel : ĠChannel < Kafka Message < Unit , ĠUnit , ĠUnit , ĠUnit , ĠWithout Transaction >>) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġrecords Ġ= Ġdelegate Mutex . with Lock Ġ{ Ġdelegate . poll ( poll Duration ) Ġ} ĊĊĠĠĠĠĠĠĠ Ġpartition Changed Messages . forEach Ġ{ Ġchannel . send ( it ) Ġ} ĊĠĠĠĠĠĠĠ Ġpartition Changed Messages . clear () ĊĠĠĠĠĠĠĠ Ġrecords . map Ġ{ ĠRecord ( it , ĠUnit , ĠUnit , ĠUnit , ĠInstant . of Epoch Milli ( it . timestamp ()), ĠUnit , ĠWithout Transaction ) Ġ}. forEach Ġ{ Ġchannel . send ( it ) Ġ} ĊĠĠĠĠĠĠĠ Ġdelegate Mutex . with Lock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ_ assignment . forEach Ġ{ Ġtopic Partition Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrun Catch ing Ġ{ Ġdelegate . position ( topic Partition ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. on Success Ġ{ Ġpositions [ topic Partition ] Ġ= Ġit Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. on Failure Ġ{ Ġpositions . remove ( topic Partition ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġif Ġ( records . isEmpty ) Ġyield () ĊĠĠĠĠĠĠĠ Ġif Ġ(! records . isEmpty ) Ġchannel . send ( EndOf Batch ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġshould Stop (): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( stop Requested ) Ġreturn Ġtrue ĊĠĠĠĠĠĠĠ Ġreturn Ġwhen Ġ( auto Stop Policy ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠAuto Stop Policy . Never Ġ-> Ġreturn Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠ ĠAuto Stop Policy . When Up ToDate Ġ-> Ġis Up ToDate () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠAuto Stop Policy . At Specific Time Ġ-> Ġauto Stop Policy . stop Time Ġ< ĠInstant . now () Ġ&& Ġis Up ToDate () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠAuto Stop Policy . Specific Offset From Now Ġ-> Ġstart Instant !! Ġ+ Ġauto Stop Policy . duration Ġ< ĠInstant . now () Ġ&& Ġis Up ToDate () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġsuspend Ġfun Ġsubscribe () Ġ{ ĊĠĠĠĠĠĠĠ Ġstart Instant Ġ= ĠInstant . now () ĊĠĠĠĠĠĠĠ Ġdelegate Mutex . with Lock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdelegate . subscribe ( topics , Ġobject Ġ: ĠConsumer Re balance Listener Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Partitions Rev oked ( partitions : ĠMutable Collection < Topic Partition >) Ġ= Ġpartition Rev oked ( partitions . toList ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Partitions Assigned ( partitions : ĠMutable Collection < Topic Partition >) Ġ= Ġpartition Assigned ( partitions . toList ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġrunning Ġ= Ġtrue ĊĠĠĠĠĠĠĠ Ġstart End Offsets Refresh Loop () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġsuspend Ġfun Ġstart End Offsets Refresh Loop () Ġ{ ĊĠĠĠĠĠĠĠ Ġend Offsets Loop Ġ= ĠCor outine Scope ( current Coroutine Context ()). launch ( Dispatch ers . IO ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġend Offset Consumer : ĠKafka Consumer < ByteArray , ĠByteArray > Ġ= ĠKafka Consumer ( properties , ĠByteArray Deserializer (), ĠByteArray Deserializer ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġend Offset Consumer . use Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ(! should Stop () Ġ&& ĠisActive ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġend Offsets Ġ= Ġend Offset Consumer . end Offsets (_ assignment ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( 1 0 . seconds (). to Millis ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( t : ĠC ancellation Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( t : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn (" Error Ġwhile Ġtrying Ġto Ġfetch Ġthe Ġend Ġoffsets ", Ġt ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġpartition Assigned ( assigned Partitions : ĠList < Topic Partition >) Ġ{ ĊĠĠĠĠĠĠĠ Ġend Offsets Ġ= Ġdelegate . end Offsets ( delegate . assignment ()) ĊĠĠĠĠĠĠĠ Ġ_ assignment Ġ= Ġdelegate . assignment (). toList () ĊĠĠĠĠĠĠĠ Ġseek (_ assignment ) ĊĠĠĠĠĠĠĠ Ġpartition Changed Messages . add ( Partitions Assigned ( assigned Partitions , Ġ_ assignment )) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġpartition Rev oked ( rev oked Partition : ĠList < Topic Partition >) Ġ{ ĊĠĠĠĠĠĠĠ Ġ_ assignment Ġ= Ġdelegate . assignment (). toList () ĊĠĠĠĠĠĠĠ Ġseek (_ assignment ) ĊĠĠĠĠĠĠĠ Ġpartition Changed Messages . add ( Partitions Rev oked ( rev oked Partition , Ġ_ assignment )) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġseek ( assigned Partitions : ĠList < Topic Partition >) Ġ{ ĊĠĠĠĠĠĠĠ Ġwhen Ġ( start Offset Policy ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠStart Offset Policy . Specific Offset From Now Ġ-> Ġseek To Specified Time ( assigned Partitions , ĠInstant . now () Ġ- Ġstart Offset Policy . duration ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠStart Offset Policy . Specific Time Ġ-> Ġseek To Specified Time ( assigned Partitions , Ġstart Offset Policy . offset Time ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠStart Offset Policy . E ar liest , ĠStart Offset Policy . Latest Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠLet Ġthe Ġkafka Ġinternal Ġclient Ġdeal Ġwith Ġthat ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġseek To Specified Time ( assigned Partitions : ĠList < Topic Partition >, Ġinstant : ĠInstant ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġend Offsets Ġ= Ġdelegate . end Offsets ( assigned Partitions ) ĊĠĠĠĠĠĠĠ Ġval Ġoffsets For Time Ġ= Ġdelegate . offsets For Times ( assigned Partitions . associate With Ġ{ Ġinstant . to Epoch Milli () Ġ}) ĊĠĠĠĠĠĠĠ Ġassigned Partitions . forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdelegate . seek ( it , Ġoffsets For Time [ it ]?. offset () Ġ?: Ġend Offsets [ it ] Ġ?: Ġ 0 ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġsuspend Ġfun Ġcleanup () Ġ{ ĊĠĠĠĠĠĠĠ Ġwith Context ( Non C ancellable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( 2 5 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdelegate . close () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrunning Ġ= Ġfalse ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġproperties ( client Properties : ĠProperties ): ĠProperties Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġproperties Ġ= ĠProperties () ĊĠĠĠĠĠĠĠ Ġproperties . put All ( client Properties ) ĊĠĠĠĠĠĠĠ Ġif Ġ( start Offset Policy Ġis ĠStart Offset Policy . E ar liest ) Ġproperties [ Consumer Config . AUTO _ OFFSET _ RESET _ CONFIG ] Ġ= Ġ" ear liest " ĊĠĠĠĠĠĠĠ Ġif Ġ( start Offset Policy Ġis ĠStart Offset Policy . Latest ) Ġproperties [ Consumer Config . AUTO _ OFFSET _ RESET _ CONFIG ] Ġ= Ġ" latest " ĊĠĠĠĠĠĠĠ Ġreturn Ġproperties ĊĠĠĠ Ġ} Ċ Ċ }