package Ġno . nav . kafka Ċ Ċ import Ġio . ktor . application . call Ċ import Ġio . ktor . application . install Ċ import Ġio . ktor . features . Content Neg otiation Ċ import Ġio . ktor . gson . gson Ċ import Ġio . ktor . http . ContentType Ċ import Ġio . ktor . response . respond Text Ċ import Ġio . ktor . response . respond Text Writer Ċ import Ġio . ktor . routing . get Ċ import Ġio . ktor . routing . routing Ċ import Ġio . ktor . server . engine . embedded Server Ċ import Ġio . ktor . server . netty . Net ty Ċ import Ġio . prometheus . client . Collector Registry Ċ import Ġio . prometheus . client . Gauge Ċ import Ġio . prometheus . client . exporter . common . Text Format Ċ import Ġio . prometheus . client . hot spot . Default Exports Ċ import Ġjava . io . File Ċ import Ġjava . util . Properties Ċ import Ġjava . util . concurrent . TimeUnit Ċ import Ġkotlin . concurrent . timer Ċ import Ġkotlinx . coroutines . run Blocking Ċ import Ġmu . Kotlin Logging Ċ import Ġorg . apache . kafka . clients . Common Client Configs Ċ import Ġorg . apache . kafka . clients . admin . Admin Client Ċ 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 . S asl Configs Ċ import Ġorg . apache . kafka . common . config . Ssl Configs Ċ Ċ private Ġval Ġlogger Ġ= ĠKotlin Logging . logger Ġ{} Ċ Ċ class ĠConsumer Offset Exporter ( environment : ĠEnvironment ) Ġ{ ĊĠĠĠ Ġprivate Ġval Ġcollector Registry : ĠCollector Registry Ġ= ĠCollector Registry . default Registry ĊĠĠĠ Ġprivate Ġval Ġkafka Credential : ĠKafka Credential Ġ= ĠKafka Credential ( environment . username , Ġenvironment . password ) ĊĠĠĠ Ġprivate Ġval Ġconsumer : ĠKafka Consumer < String , ĠString > Ġ= Ġcreate New Consumer ( environment ) ĊĠĠĠ Ġprivate Ġval Ġadmin Client : ĠAdmin Client Ġ= Ġcreate Admin Client ( environment ) ĊĠĠĠ Ġprivate Ġval Ġhttp Port : ĠInt Ġ= Ġ 8 0 8 0 ĊĠĠĠ Ġprivate Ġval Ġconsumer Groups : ĠString Ġ= Ġenvironment . consumer Groups ĊĊĠĠĠ Ġprivate Ġval Ġoffset Lag Gauge : ĠG auge Ġ= ĠG auge . build () ĊĠĠĠĠĠĠĠ Ġ. namespace ( environment . namespace ) ĊĠĠĠĠĠĠĠ Ġ. name (" consumer _ offset _ lag ") ĊĠĠĠĠĠĠĠ Ġ. help (" Offset Ġlag Ġof Ġa Ġtopic / partition ") ĊĠĠĠĠĠĠĠ Ġ. label Names (" group _ id ", Ġ" partition ", Ġ" topic ") ĊĠĠĠĠĠĠĠ Ġ. register () ĊĊĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠ ĠDefault Exports . initialize () ĊĠĠĠĠĠĠĠ ĠRuntime . get Runtime (). add Shutdown Hook ( Thread Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" Closing Ġthe Ġapplication ...") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer . close () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" done !") ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠ Ġlogger . info (" Monitoring Ġconsumer Ġgroup ( s ) Ġ${ consumer Groups . split ("," ). join ToString Ġ{ Ġ"\" $ it \"\ n " Ġ}} ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġ@ Jvm Static ĊĠĠĠĠĠĠĠ Ġfun Ġmain ( args : ĠArray < String >) Ġ= Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġoffset Exporter Ġ= ĠConsumer Offset Exporter ( Environment ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġoffset Exporter . start () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġkafka Offset Sc raper () Ġ{ ĊĠĠĠĠĠĠĠ Ġconsumer Groups . split ("," ). forEach Ġ{ Ġgroup Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġconsumer Group Offsets Ġ= Ġadmin Client . list Consumer Group Offsets ( group ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer Group Offsets . partitions To Offset And Metadata (). when Complete Ġ{ Ġtopic Partitions Offsets , Ġthrowable Ġ-> ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtopic Partitions Offsets ?. forEach Ġ{ Ġtopic Partition , Ġoffset Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġcurrent Offset Ġ= Ġoffset . offset () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġend Offset Ġ= Ġget Log End Offset ( topic Partition ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġlag Ġ= Ġend Offset Ġ- Ġcurrent Offset ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoffset Lag Gauge . labels ( group , Ġtopic Partition . partition (). toString (), Ġtopic Partition . topic ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. set ( lag . toDouble ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug (" Lag Ġis Ġ-> Ġ$ lag Ġfor Ġtopic Ġ'${ topic Partition . topic ()} ', Ġpartition Ġ${ topic Partition . partition ()}, Ġcurrent Ġoffset Ġ$ current Offset Ġ, Ġend Ġoffset Ġ$ end Offset Ġfor Ġgroup : Ġ$ group ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrowable ?. apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . warn ( throwable ) Ġ{ Ġ" Failed Ġto Ġget Ġoffset Ġdata Ġfrom Ġconsumer Ġgroup Ġ$ group " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcreate Admin Client ( environment : ĠEnvironment ): ĠAdmin Client Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġprops Ġ= ĠProperties (). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Consumer Config . BOOT STR AP _ SERVER S _ CONFIG , Ġenvironment . bootstrap Servers Url ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput All ( credential Properties ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn ĠAdmin Client . create ( props ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġsuspend Ġfun Ġstart () Ġ{ ĊĠĠĠĠĠĠĠ Ġlogger . info Ġ{ Ġ" START ING " Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġtimer Task Ġ= Ġtimer ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġname Ġ= Ġ" offset - checker - task ", ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdaemon Ġ= Ġtrue , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġinitial Delay Ġ= ĠTimeUnit . SECONDS . to Millis ( 5 ), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġperiod Ġ= ĠTimeUnit . SECONDS . to Millis ( 1 0 ) ĊĠĠĠĠĠĠĠ Ġ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . debug (" Loop ing Ġin Ġtimer Ġtask ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġkafka Offset Sc raper () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġapp Ġ= Ġembedded Server ( Net ty , Ġhttp Port ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġinstall ( Content Neg otiation ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġgson Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġset Pretty Print ing () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrouting Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġget ("/ is Alive ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcall . respond Text (" AL IVE ", ĠContentType . Text . Plain ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġget ("/ is Ready ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcall . respond Text (" READY ", ĠContentType . Text . Plain ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġget ("/ metrics ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġnames Ġ= Ġcall . request . query Parameters . getAll (" name ") ?. toSet () Ġ?: Ġset Of () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcall . respond Text Writer ( ContentType . parse ( Text Format . CONTENT _ TYPE _ 0 0 4 )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠText Format . write 0 0 4 ( this , Ġcollector Registry . filtered Metric Family Samples ( names )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ}. start ( wait Ġ= Ġfalse ) ĊĊĠĠĠĠĠĠĠ ĠRuntime . get Runtime (). add Shutdown Hook ( Thread Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġapp . stop ( 3 0 0 0 , Ġ 5 0 0 0 ) ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠ ĠRuntime . get Runtime (). add Shutdown Hook ( Thread Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtimer Task . cancel () ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġget Log End Offset ( topic Partition : ĠTopic Partition ): ĠLong Ġ{ ĊĠĠĠĠĠĠĠ Ġconsumer . assign ( listOf ( topic Partition )) ĊĠĠĠĠĠĠĠ Ġconsumer . seek ToEnd ( listOf ( topic Partition )) ĊĠĠĠĠĠĠĠ Ġreturn Ġconsumer . position ( topic Partition ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcreate New Consumer ( environment : ĠEnvironment ): ĠKafka Consumer < String , ĠString > Ġ{ ĊĠĠĠĠĠĠĠ Ġval ĠgroupId Ġ= Ġenvironment . namespace Ġ+ Ġ"- offset checker " ĊĊĠĠĠĠĠĠĠ Ġval Ġproperties Ġ= ĠProperties (). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Consumer Config . GROUP _ ID _ CONFIG , ĠgroupId ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Consumer Config . CLIENT _ ID _ CONFIG , ĠgroupId ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Consumer Config . BOOT STR AP _ SERVER S _ CONFIG , Ġenvironment . bootstrap Servers Url ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Consumer Config . ENABLE _ AUTO _ COMMIT _ CONFIG , Ġ" false ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Consumer Config . SESSION _ TIMEOUT _ MS _ CONFIG , Ġ" 3 0 0 0 0 ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠConsumer Config . KEY _ DE SERIAL IZER _ CLASS _ CONFIG , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" org . apache . kafka . common . serialization . String Deserializer " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠConsumer Config . VALUE _ DE SERIAL IZER _ CLASS _ CONFIG , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" org . apache . kafka . common . serialization . String Deserializer " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput All ( credential Properties ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn ĠKafka Consumer ( properties ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcredential Properties (): ĠProperties Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠProperties (). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġkafka Credential . let Ġ{ Ġcredential Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info Ġ{ Ġ" Using Ġuser Ġname Ġ${ credential . username } Ġto Ġauthenticate Ġagainst ĠKafka Ġbro kers Ġ" Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput ( S asl Configs . SAS L _ ME CHAN IS M , Ġ" PLAIN ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Common Client Configs . SECURITY _ PROTOCOL _ CONFIG , Ġ" SAS L _ PL AINT EXT ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠS asl Configs . SAS L _ JA AS _ CONFIG , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" org . apache . kafka . common . security . plain . Plain Login Module Ġrequired Ġusername =\" ${ credential . username }\" Ġpassword =\" ${ credential . password }\ "; " ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtrust Store Location Ġ= ĠSystem . getenv (" NAV _ TR UST STORE _ PATH ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtrust Store Location ?. let Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Common Client Configs . SECURITY _ PROTOCOL _ CONFIG , Ġ" SAS L _ SSL ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Ssl Configs . SSL _ TR UST STORE _ LOCATION _ CONFIG , ĠFile ( it ). absolute Path ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Ssl Configs . SSL _ TR UST STORE _ PASSWORD _ CONFIG , ĠSystem . getenv (" NAV _ TR UST STORE _ PASSWORD ")) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info Ġ{ Ġ" Configured Ġ'${ Ssl Configs . SSL _ TR UST STORE _ LOCATION _ CONFIG }' Ġlocation Ġ" Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error Ġ{ Ġ" Failed Ġto Ġset Ġ'${ Ssl Configs . SSL _ TR UST STORE _ LOCATION _ CONFIG }' Ġlocation Ġ" Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġdata Ġclass ĠKafka Credential ( val Ġusername : ĠString , Ġval Ġpassword : ĠString ) Ġ{ ĊĠĠĠĠĠĠĠ Ġoverride Ġfun ĠtoString (): ĠString Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġ" username Ġ'$ username ' Ġpassword Ġ' ******* '" ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ