package Ġno . nav . sy fo . et ter level se Ċ Ċ import Ġkotlinx . coroutines . Del icate Cor outines Api Ċ import Ġkotlinx . coroutines . Dispatch ers Ċ import Ġkotlinx . coroutines . Global Scope Ċ import Ġkotlinx . coroutines . delay Ċ import Ġkotlinx . coroutines . launch Ċ import Ġno . nav . sy fo . application . Application State Ċ import Ġno . nav . sy fo . application . metrics . PRO DU CED _ MESSAGE _ COUNTER Ċ import Ġno . nav . sy fo . et ter level se . model . J ur id isk V ur der ing Kafka Message Ċ import Ġno . nav . sy fo . et ter level se . model . J ur id isk V ur der ing Result Ċ import Ġno . nav . sy fo . log Ċ import Ġno . nav . sy fo . util . Un bounded Ċ import Ġorg . apache . kafka . clients . consumer . Kafka Consumer Ċ import Ġorg . apache . kafka . clients . producer . Kafka Producer Ċ import Ġorg . apache . kafka . clients . producer . Producer Record Ċ import Ġjava . time . Duration Ċ import Ġjava . time . Instant Ċ Ċ class ĠEt ter level se Service ( ĊĠĠĠ Ġprivate Ġval Ġapplication State : ĠApplication State , ĊĠĠĠ Ġprivate Ġval Ġkafka Consumer : ĠKafka Consumer < String , ĠJ ur id isk V ur der ing Result >, ĊĠĠĠ Ġprivate Ġval Ġintern P ik Topic : ĠString , ĊĠĠĠ Ġprivate Ġval Ġkafka Producer : ĠKafka Producer < String , ĠJ ur id isk V ur der ing Kafka Message >, ĊĠĠĠ Ġprivate Ġval Ġet ter level se Topic : ĠString Ċ ) Ġ{ ĊĠĠĠ Ġprivate Ġvar Ġlast Log Time Ġ= ĠInstant . now (). to Epoch Milli () ĊĠĠĠ Ġprivate Ġval Ġlog Timer Ġ= Ġ 6 0 _ 0 0 0 L ĊĊĠĠĠ Ġ@ Del icate Cor outines Api ĊĠĠĠ Ġfun Ġstart Consumer () Ġ{ ĊĠĠĠĠĠĠĠ ĠGlobal Scope . launch ( Dispatch ers . Un bounded ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( application State . ready ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info (" Starting Ġconsuming Ġtopic ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġkafka Consumer . subscribe ( listOf ( intern P ik Topic )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstart () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . error (" Error Ġrunning Ġkafka Ġconsumer , Ġun sub scri bing Ġand Ġwaiting Ġ 1 0 Ġseconds Ġfor Ġretry ", Ġex ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġkafka Consumer . unsubscribe () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( 1 0 _ 0 0 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġstart () Ġ{ ĊĠĠĠĠĠĠĠ Ġvar Ġprocessed Messages Ġ= Ġ 0 ĊĠĠĠĠĠĠĠ Ġwhile Ġ( application State . ready ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġrecords Ġ= Ġkafka Consumer . poll ( Duration . of Seconds ( 1 0 )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrecords . forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . value (). j ur id is ke V ur der inger . forEach Ġ{ Ġj ur id isk V ur der ing Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsend To Kafka ( j ur id isk V ur der ing . til J ur id isk V ur der ing Kafka Message ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprocessed Messages Ġ+= Ġrecords . count () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprocessed Messages Ġ= Ġlog Processed Messages ( processed Messages ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġsend To Kafka ( j ur id isk V ur der ing Kafka Message : ĠJ ur id isk V ur der ing Kafka Message ) Ġ{ ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġkafka Producer . send ( Producer Record ( et ter level se Topic , Ġj ur id isk V ur der ing Kafka Message . f od se ls nummer , Ġj ur id isk V ur der ing Kafka Message )). get () ĊĠĠĠĠĠĠĠĠĠĠĠ ĠPRO DU CED _ MESSAGE _ COUNTER . inc () ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . error (" Failed Ġto Ġsend Ġmessage Ġto Ġkafka Ġfor Ġid Ġ${ j ur id isk V ur der ing Kafka Message . id }") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġex ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġlog Processed Messages ( processed Messages : ĠInt ): ĠInt Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġcurrent Log Time Ġ= ĠInstant . now (). to Epoch Milli () ĊĠĠĠĠĠĠĠ Ġif Ġ( processed Messages Ġ> Ġ 0 Ġ&& Ġcurrent Log Time Ġ- Ġlast Log Time Ġ> Ġlog Timer ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info (" Processed Ġ$ processed Messages Ġmessages ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlast Log Time Ġ= Ġcurrent Log Time ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġ 0 ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġprocessed Messages ĊĠĠĠ Ġ} Ċ } Ċ