package Ġno . nav . ar be id sg iver . toi Ċ Ċ import Ġkotlinx . coroutines . Global Scope Ċ import Ġkotlinx . coroutines . Job Ċ import Ġkotlinx . coroutines . launch Ċ import Ġkotlinx . coroutines . time . delay Ċ import Ġno . nav . hel se . rap ids _ river s . R apid Application Ċ import Ġno . nav . hel se . rap ids _ river s . R ap ids Connection Ċ import Ġorg . apache . kafka . clients . Common Client Configs Ċ import Ġorg . apache . kafka . clients . consumer . Consumer Config Ċ import Ġorg . apache . kafka . clients . consumer . Kafka Consumer Ċ import Ġorg . apache . kafka . common . Topic Partition Ċ import Ġorg . apache . kafka . common . config . Ssl Configs Ċ import Ġorg . apache . kafka . common . security . auth . Security Protocol Ċ import Ġorg . apache . kafka . common . serialization . String Deserializer Ċ import Ġorg . slf 4 j . Logger Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġjava . time . Duration Ċ import Ġjava . util .* Ċ Ċ fun Ġmain () Ġ{ ĊĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠ Ġvar Ġoffset Job : ĠJob ? Ġ= Ġnull ĊĠĠĠĠĠĠĠ Ġval Ġenv s Ġ= ĠSystem . getenv () ĊĠĠĠĠĠĠĠ ĠR apid Application . create ( envs ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġregister ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġobject Ġ: ĠR ap ids Connection . Status Listener Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Startup ( rap ids Connection : ĠR ap ids Connection ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoffset Job Ġ= ĠGlobal Scope . launch Ġ{ Ġs je kk Offsets ( envs ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoffset Job ?. invoke On Completion Ġ{ Ġrap ids Connection . stop () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Shutdown ( rap ids Connection : ĠR ap ids Connection ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoffset Job ?. cancel () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. start () ĊĊĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠ Ġlog . error ( e . message , Ġe ) ĊĠĠĠ Ġ} Ċ } Ċ Ċ suspend Ġfun Ġs je kk Offsets ( envs : ĠMap < String , ĠString >) Ġ{ ĊĠĠĠ Ġval ĠlistOf Group Ids Ġ= ĠlistOf ( ĊĠĠĠĠĠĠĠ Ġ" toi - cv " Ġto Ġ" toi - ar be id sp lass en - cv - reader - rapid consumer - 1 ", ĊĠĠĠĠĠĠĠ Ġ" toi - f rit att - k andid ats Ã¸ k " Ġto Ġ" toi - arena - cv - reader - rapid consumer - 1 ", ĊĠĠĠĠĠĠĠ Ġ" toi - hj em mel " Ġto Ġ" toi - hj em mel - rapid consumer - 1 ", ĊĠĠĠĠĠĠĠ Ġ" toi - ident mapper " Ġto Ġ" toi - ident mapper - rapid consumer - 4 ", ĊĠĠĠĠĠĠĠ Ġ" toi - k andid at feed " Ġto Ġ" toi - k andid at feed - rapid consumer - 6 ", ĊĠĠĠĠĠĠĠ Ġ" toi - ma a - be handle - tid lig ere - cv " Ġto Ġ" toi - ma a - be handle - tid lig ere - cv - rapid consumer - 4 ", ĊĠĠĠĠĠĠĠ Ġ" toi - op pf olg ings inform as jon " Ġto Ġ" toi - op pf Ã¸ lg ings inform as jon - rapid consumer - 3 ", ĊĠĠĠĠĠĠĠ Ġ" toi - op pf ol ging sp eri ode " Ġto Ġ" toi - op pf Ã¸ l ging sp eri ode - rapid consumer - 1 ", ĊĠĠĠĠĠĠĠ Ġ" toi - organ is as j ons en het " Ġto Ġ" toi - organ is as j ons en het - rapid consumer - 1 ", ĊĠĠĠĠĠĠĠ Ġ" toi - sam men st ille - k andid at " Ġto Ġ" toi - sam men st ille - k andid at - rapid consumer - 1 ", ĊĠĠĠĠĠĠĠ Ġ" toi - syn lig hets motor " Ġto Ġ" toi - syn lig hets motor - rapid consumer - 4 ", ĊĠĠĠĠĠĠĠ Ġ" toi - til ret te leg g ings be h ov " Ġto Ġ" toi - til ret te leg g ings be h ov - reader - rapid consumer - 1 ", ĊĠĠĠĠĠĠĠ Ġ" toi - ve ile der " Ġto Ġ" toi - ve ile der - rapid consumer - 9 ", ĊĠĠĠĠĠĠĠ Ġ" toi - hull - i - cv " Ġto Ġ" toi - hull - i - cv - rapid consumer - 1 " ĊĠĠĠ Ġ) ĊĠĠĠ Ġwhile Ġ( true ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġsist e Offset Ġ= Ġsist e Offset ( envs ) ĊĠĠĠĠĠĠĠ Ġval Ġresults Per Application Ġ= ĠlistOf Group Ids . map Ġ{ Ġ( application , ĠgroupId ) Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġconsumer Offset Ġ= Ġconsumer Offset ( groupId , Ġenv s ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠResults Per Application ( application , Ġconsumer Offset , Ġsist e Offset Ġ- Ġconsumer Offset ) ĊĠĠĠĠĠĠĠ Ġ}. sorted By Descending ( Results Per Application :: behind ) ĊĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġformat Results ( results Per Application , Ġsist e Offset ) ĊĠĠĠĠĠĠĠ Ġlog . info ( result ) ĊĠĠĠĠĠĠĠ Ġdelay ( Duration . of Seconds ( 1 0 )) ĊĠĠĠ Ġ} Ċ } Ċ Ċ private Ġfun Ġformat Results ( ĊĠĠĠ Ġresults Per Application : ĠList < Results Per Application >, ĊĠĠĠ Ġsist e Offset : ĠLong Ċ ) Ġ= Ġ""" ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ${ results Per Application . join ToString ( separator Ġ= Ġ"\ n ", transform Ġ= ĠResults Per Application :: print Friendly )} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠS iste Ġoffset Ġer Ġ$ s iste Offset Ċ """. trim Indent () Ċ Ċ data Ġclass ĠResults Per Application ( val Ġname : ĠString , Ġval Ġoffset : ĠLong , Ġval Ġbehind : ĠLong ) Ġ{ ĊĠĠĠ Ġfun Ġprint Friendly () Ġ= Ġ"$ name Ġis Ġon Ġoffset : Ġ$ offset Ġ( Ġ$ behind Ġbehind Ġlast Ġoffset )" Ċ } Ċ Ċ fun Ġconsumer Offset ( groupId : ĠString , Ġenv s : ĠMap < String , ĠString >): ĠLong Ġ{ ĊĠĠĠ Ġval Ġtopic Partition Ġ= ĠTopic Partition ( envs [" K AF KA _ R AP ID _ TOPIC "], Ġ 0 ) ĊĠĠĠ Ġval Ġconsumer Ġ= ĠKafka Consumer ( consumer Properties ( envs , ĠgroupId ), ĠString Deserializer (), ĠString Deserializer ()) ĊĠĠĠ Ġval Ġres Ġ= Ġ( consumer . committed ( set Of ( topic Partition ))[ topic Partition ]?. offset () ĊĠĠĠĠĠĠĠ Ġ?: Ġthrow ĠException (" F ant Ġing en Ġoffset Ġfor ĠgroupId : Ġ$ groupId ")) ĊĠĠĠ Ġconsumer . close () ĊĠĠĠ Ġreturn Ġres Ċ } Ċ Ċ fun Ġsist e Offset ( envs : ĠMap < String , ĠString >): ĠLong Ġ{ ĊĠĠĠ Ġval Ġkafka Consumer Ġ= ĠKafka Consumer ( ĊĠĠĠĠĠĠĠ Ġconsumer Properties ( envs , Ġenv s [" K AF KA _ CON SUM ER _ GROUP _ ID "] !! , ĠclientId Ġ= Ġ" toi - hel se app "), ĊĠĠĠĠĠĠĠ ĠString Deserializer (), ĊĠĠĠĠĠĠĠ ĠString Deserializer () ĊĠĠĠ Ġ) ĊĊĠĠĠ Ġval Ġtopic Partition Ġ= ĠTopic Partition ( envs [" K AF KA _ R AP ID _ TOPIC "], Ġ 0 ) ĊĠĠĠ Ġkafka Consumer . assign ( listOf ( topic Partition )) ĊĠĠĠ Ġkafka Consumer . seek ToEnd ( listOf ( topic Partition )) ĊĠĠĠ Ġval Ġposition Ġ= Ġkafka Consumer . position ( topic Partition ) ĊĠĠĠ Ġkafka Consumer . close () ĊĠĠĠ Ġreturn Ġposition Ċ } Ċ Ċ private Ġfun Ġconsumer Properties ( envs : ĠMap < String , ĠString >, ĠgroupId : ĠString , ĠclientId : ĠString Ġ= Ġ" consumer - toi - hel se app -$ groupId ") Ġ= ĠProperties (). apply Ġ{ ĊĠĠĠ Ġput ( Common Client Configs . BOOT STR AP _ SERVER S _ CONFIG , Ġenv s [" K AF KA _ BRO K ERS "]) ĊĠĠĠ Ġput ( Common Client Configs . SECURITY _ PROTOCOL _ CONFIG , ĠSecurity Protocol . SSL . name ) ĊĠĠĠ Ġput ( Ssl Configs . SSL _ ENDPOINT _ IDENT IFICATION _ ALGORITHM _ CONFIG , Ġ"") ĊĠĠĠ Ġput ( Ssl Configs . SSL _ TR UST STORE _ TYPE _ CONFIG , Ġ" j ks ") ĊĠĠĠ Ġput ( Ssl Configs . SSL _ KEY STORE _ TYPE _ CONFIG , Ġ" PKCS 1 2 ") ĊĠĠĠ Ġput ( Ssl Configs . SSL _ TR UST STORE _ LOCATION _ CONFIG , Ġenv s [" K AF KA _ TR UST STORE _ PATH "]) ĊĠĠĠ Ġput ( Ssl Configs . SSL _ TR UST STORE _ PASSWORD _ CONFIG , Ġenv s [" K AF KA _ CRED STORE _ PASSWORD "]) ĊĠĠĠ Ġput ( Ssl Configs . SSL _ KEY STORE _ LOCATION _ CONFIG , Ġenv s [" K AF KA _ KEY STORE _ PATH "]) ĊĠĠĠ Ġput ( Ssl Configs . SSL _ KEY STORE _ PASSWORD _ CONFIG , Ġenv s [" K AF KA _ CRED STORE _ PASSWORD "]) ĊĠĠĠ Ġput ( Consumer Config . GROUP _ ID _ CONFIG , ĠgroupId ) ĊĠĠĠ Ġput ( Consumer Config . CLIENT _ ID _ CONFIG , ĠclientId ) ĊĠĠĠ Ġput ( Consumer Config . ENABLE _ AUTO _ COMMIT _ CONFIG , Ġfalse ) ĊĠĠĠ Ġput ( Consumer Config . AUTO _ OFFSET _ RESET _ CONFIG , Ġ" latest ") ĊĠĠĠ Ġput ( Consumer Config . MAX _ POLL _ RECORD S _ CONFIG , Ġ" 2 0 0 ") ĊĠĠĠ Ġput ( Consumer Config . MAX _ POLL _ INTERVAL _ MS _ CONFIG , Ġ"${ Duration . of Seconds ( 6 0 Ġ+ Ġ 2 0 0 Ġ* Ġ 2 . to Long ()). to Millis () }") Ċ } Ċ Ċ val Ġlog : ĠLogger ĊĠĠĠ Ġget () Ġ= ĠLoggerFactory . getLogger (" toi - hel se app ") Ċ