package Ġno . nav . so kn ad . ark iver ing . kafka Ċ Ċ import Ġcom . fasterxml . jackson . databind . ObjectMapper Ċ import Ġcom . fasterxml . jackson . databind . Serialization Feature Ċ import Ġcom . fasterxml . jackson . module . kotlin . readValue Ċ import Ġio . con fluent . kafka . serializers . Abstract Kafka Schema Ser De Config Ċ import Ġio . con fluent . kafka . streams . ser des . avro . Specific Av ro Ser de Ċ import Ġno . nav . so kn ad . ark iver ing . Configuration Ċ import Ġno . nav . so kn ad . ark iver ing . avro schemas . In ns ending Metrics Ċ import Ġno . nav . so kn ad . ark iver ing . avro schemas . Processing Event Ċ import Ġno . nav . so kn ad . ark iver ing . dto . Archive Entity Ċ import Ġorg . apache . avro . specific . Specific Record Ċ import Ġorg . apache . kafka . clients . Common Client Configs Ċ import Ġorg . apache . kafka . common . config . S asl Configs Ċ import Ġorg . apache . kafka . common . serialization . Ser des Ċ import Ġorg . apache . kafka . streams . Kafka Streams Ċ import Ġorg . apache . kafka . streams . KeyValue Ċ import Ġorg . apache . kafka . streams . Streams Builder Ċ import Ġorg . apache . kafka . streams . Streams Config Ċ import Ġorg . apache . kafka . streams . errors . Log And Continue ExceptionHandler Ċ import Ġorg . apache . kafka . streams . k stream . Cons umed Ċ import Ġorg . apache . kafka . streams . k stream . Transformer Ċ import Ġorg . apache . kafka . streams . processor . Processor Context Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġjava . util .* Ċ Ċ class ĠKafka Listener ( private Ġval Ġkafka Config : ĠConfiguration . Kafka Config ) Ġ{ ĊĊ ĉ private Ġval Ġlogger Ġ= ĠLoggerFactory . getLogger ( java Class ) Ċ ĉ private Ġval Ġverbose Ġ= Ġtrue ĊĊ ĉ private Ġval Ġentity Consum ers ĠĠĠĠĠĠĠĠĠ Ġ= Ġmutable ListOf < Kafka Entity Consumer < Archive Entity >>() Ċ ĉ private Ġval Ġmetrics Consum ers ĠĠĠĠĠĠĠĠ Ġ= Ġmutable ListOf < Kafka Entity Consumer < In ns ending Metrics >>() Ċ ĉ private Ġval ĠnumberOf Calls Consum ers ĠĠ Ġ= Ġmutable ListOf < Kafka Entity Consumer < Int >>() Ċ ĉ private Ġval Ġprocessing Event Consum ers Ġ= Ġmutable ListOf < Kafka Entity Consumer < Processing Event >>() ĊĊ ĉ private Ġval Ġkafka Streams : ĠKafka Streams ĊĊ ĉ private Ġval Ġint Ser de Ġ= ĠSer des . Integer Ser de () Ċ ĉ private Ġval Ġstring Ser de Ġ= ĠSer des . String Ser de () Ċ ĉ private Ġval Ġmapper : ĠObjectMapper Ġ= ĠObjectMapper (). also Ġ{ Ċĉ ĉ it . enable ( Serialization Feature . IND ENT _ OUTPUT ) Ċĉ ĉ it . find And Register Modules () Ċ ĉ } ĊĊ ĉ init Ġ{ Ċĉ ĉ val Ġstreams Builder Ġ= ĠStreams Builder () Ċĉ ĉ kafka Streams ( streams Builder ) Ċĉ ĉ val Ġtopology Ġ= Ġstreams Builder . build () ĊĊĉ ĉ val Ġkafka Config Ġ= Ġkafka Config () Ċĉ ĉ kafka Streams Ġ= ĠKafka Streams ( topology , Ġkafka Config ) Ċĉ ĉ kafka Streams . start () Ċĉ ĉ Runtime . get Runtime (). add Shutdown Hook ( Thread ( kafka Streams :: close )) Ċ ĉ } ĊĊĊ ĉ private Ġfun Ġkafka Streams ( streams Builder : ĠStreams Builder ) Ġ{ Ċĉ ĉ val Ġmetrics Stream ĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ= Ġstreams Builder . stream ( kafka Config . metrics Topic , ĠĠĠĠĠĠ ĠCon su med . with ( string Ser de , Ġcreate In ns ending Metrics Ser de ())) Ċĉ ĉ val Ġprocessing Event Topic Stream Ġ= Ġstreams Builder . stream ( kafka Config . processing Topic , ĠĠĠ ĠCon su med . with ( string Ser de , Ġcreate Processing Event Ser de ())) Ċĉ ĉ val Ġentities Stream ĠĠĠĠĠĠĠĠĠĠĠĠ Ġ= Ġstreams Builder . stream ( kafka Config . entities Topic , ĠĠĠĠĠ ĠCon su med . with ( string Ser de , Ġstring Ser de )) Ċĉ ĉ val ĠnumberOf Calls Stream ĠĠĠĠĠĠĠ Ġ= Ġstreams Builder . stream ( kafka Config . numberOf Calls Topic , ĠCon su med . with ( string Ser de , Ġint Ser de )) ĊĊĉ ĉ entities Stream Ċĉĉ ĉ . map Values Ġ{ Ġjson Ġ-> Ġmapper . readValue < Archive Entity >( json ) Ġ} Ċĉĉ ĉ . peek Ġ{ Ġkey , Ġentity Ġ-> Ġlog ("$ key : ĠArchive ĠEntities Ġ Ġ- Ġ$ entity ") Ġ} Ċĉĉ ĉ . transform ({ ĠTimestamp Extractor () Ġ}) Ċĉĉ ĉ . foreach Ġ{ Ġkey , Ġentity Ġ-> Ġentity Consum ers . forEach Ġ{ Ġit . consume ( key , Ġentity ) Ġ} Ġ} ĊĊĉ ĉ metrics Stream Ċĉĉ ĉ . peek Ġ{ Ġkey , Ġentity Ġ-> Ġlog ("$ key : ĠMetrics Ġreceived Ġ Ġ- Ġ$ entity ") Ġ} Ċĉĉ ĉ . transform ({ ĠTimestamp Extractor () Ġ}) Ċĉĉ ĉ . foreach Ġ{ Ġkey , Ġentity Ġ-> Ġmetrics Consum ers . forEach Ġ{ Ġit . consume ( key , Ġentity ) Ġ} Ġ} ĊĊĉ ĉ numberOf Calls Stream Ċĉĉ ĉ . peek Ġ{ Ġkey , ĠnumberOf Calls Ġ-> Ġlog ("$ key : ĠNumber Ġof ĠCalls ĠĠ Ġ- Ġ$ numberOf Calls ") Ġ} Ċĉĉ ĉ . transform ({ ĠTimestamp Extractor () Ġ}) Ċĉĉ ĉ . foreach Ġ{ Ġkey , ĠnumberOf Calls Ġ-> ĠnumberOf Calls Consum ers . forEach Ġ{ Ġit . consume ( key , ĠnumberOf Calls ) Ġ} Ġ} ĊĊĉ ĉ processing Event Topic Stream Ċĉĉ ĉ . peek Ġ{ Ġkey , Ġentity Ġ-> Ġlog ("$ key : ĠProcessing ĠEvents Ġ- Ġ$ entity ") Ġ} Ċĉĉ ĉ . transform ({ ĠTimestamp Extractor () Ġ}) Ċĉĉ ĉ . foreach Ġ{ Ġkey , Ġentity Ġ-> Ġprocessing Event Consum ers . forEach Ġ{ Ġit . consume ( key , Ġentity ) Ġ} Ġ} Ċ ĉ } ĊĊ ĉ private Ġfun Ġlog ( message : ĠString ) Ġ{ Ċĉ ĉ if Ġ( verbose ) Ċĉĉ ĉ logger . info ( message ) Ċ ĉ } ĊĊ ĉ private Ġfun Ġkafka Config () Ġ= ĠProperties (). also Ġ{ Ċĉ ĉ it [ Abstract Kafka Schema Ser De Config . SCHEMA _ REGISTRY _ URL _ CONFIG ] Ġ= Ġkafka Config . schema Registry Url Ċĉ ĉ it [ Streams Config . APPLICATION _ ID _ CONFIG ] Ġ= Ġ" in ns ending - system - tests " Ċĉ ĉ it [ Streams Config . BOOT STR AP _ SERVER S _ CONFIG ] Ġ= Ġkafka Config . servers Ċĉ ĉ it [ Streams Config . DEFAULT _ KEY _ SER DE _ CLASS _ CONFIG ] Ġ= ĠSer des . String Ser de :: class . java Ċĉ ĉ it [ Streams Config . DEFAULT _ VALUE _ SER DE _ CLASS _ CONFIG ] Ġ= ĠSpecific Av ro Ser de :: class . java Ċĉ ĉ it [ Streams Config . DEFAULT _ DE SERIAL IZATION _ EXCEPTION _ HANDLER _ CLASS _ CONFIG ] Ġ= ĠLog And Continue ExceptionHandler :: class . java Ċĉ ĉ it [ Streams Config . COMMIT _ INTERVAL _ MS _ CONFIG ] Ġ= Ġ 1 0 0 0 ĊĊĉ ĉ if Ġ( kafka Config . secure Ġ== Ġ" TRUE ") Ġ{ Ċĉĉ ĉ it [ Common Client Configs . SECURITY _ PROTOCOL _ CONFIG ] Ġ= Ġkafka Config . protocol Ċĉĉ ĉ it [ S asl Configs . SAS L _ JA AS _ CONFIG ] Ġ= Ġkafka Config . sas l Ja as Config Ċĉĉ ĉ it [ S asl Configs . SAS L _ ME CHAN IS M ] Ġ= Ġkafka Config . s als me c Ċĉ ĉ } Ċ ĉ } ĊĊ ĉ private Ġfun Ġcreate Processing Event Ser de (): ĠSpecific Av ro Ser de < Processing Event > Ġ= Ġcreate Av ro Ser de () Ċ ĉ private Ġfun Ġcreate In ns ending Metrics Ser de (): ĠSpecific Av ro Ser de < In ns ending Metrics > Ġ= Ġcreate Av ro Ser de () ĊĊ ĉ private Ġfun Ġ< T Ġ: ĠSpecific Record > Ġcreate Av ro Ser de (): ĠSpecific Av ro Ser de < T > Ġ{ Ċĉ ĉ val Ġserde Config Ġ= Ċĉĉ ĉ hash Map Of ( Abstract Kafka Schema Ser De Config . SCHEMA _ REGISTRY _ URL _ CONFIG Ġto Ġkafka Config . schema Registry Url ) Ċĉ ĉ return ĠSpecific Av ro Ser de < T >(). also Ġ{ Ġit . configure ( serde Config , Ġfalse ) Ġ} Ċ ĉ } ĊĊ ĉ /** Ċĉ Ġ* ĠThis Ġclass Ġis Ġjust Ġboilerplate Ġfor Ġextracting Ġa Ġtimestamp Ġfrom Ġa ĠKafka Ġrecord . Ċĉ Ġ*/ Ċ ĉ class ĠTimestamp Extractor < T > Ġ: ĠTransformer < String , ĠT , ĠKey Value < String , ĠKafka Timestamp ed Entity < T >>> Ġ{ Ċĉ ĉ private Ġlateinit Ġvar Ġcontext : ĠProcessor Context Ċĉ ĉ override Ġfun Ġinit ( context : ĠProcessor Context ) Ġ{ Ċĉĉ ĉ this . context Ġ= Ġcontext Ċĉ ĉ } ĊĊĉ ĉ override Ġfun Ġtransform ( key : ĠString , Ġvalue : ĠT ): ĠKey Value < String , ĠKafka Timestamp ed Entity < T >> Ġ{ Ċĉĉ ĉ return ĠKey Value ( key , ĠKafka Timestamp ed Entity ( value , Ġcontext . timestamp ())) Ċĉ ĉ } ĊĊĉ ĉ override Ġfun Ġclose () Ġ{ Ċĉ ĉ } Ċ ĉ } ĊĊĊ ĉ fun Ġclose () Ġ{ Ċĉ ĉ kafka Streams . close () Ċĉ ĉ kafka Streams . clean Up () Ċ ĉ } ĊĊĊ ĉ fun Ġclear Consum ers () Ġ{ Ċĉ ĉ entity Consum ers . clear () Ċĉ ĉ metrics Consum ers . clear () Ċĉ ĉ numberOf Calls Consum ers . clear () Ċĉ ĉ processing Event Consum ers . clear () Ċ ĉ } ĊĊ ĉ @ Suppress (" unused ") Ċ ĉ fun Ġadd Consumer For Metrics ĠĠĠĠĠĠĠĠ Ġ( consumer : ĠKafka Entity Consumer < In ns ending Metrics >) Ġ= Ġmetrics Consum ers ĠĠĠĠĠĠĠ Ġ. add ( consumer ) Ċ ĉ fun Ġadd Consumer For Entities ĠĠĠĠĠĠĠ Ġ( consumer : ĠKafka Entity Consumer < Archive Entity >) ĠĠĠĠĠĠ Ġ= Ġentity Consum ers ĠĠĠĠĠĠĠĠ Ġ. add ( consumer ) Ċ ĉ fun Ġadd Consumer For NumberOf Calls ĠĠ Ġ( consumer : ĠKafka Entity Consumer < Int >) ĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ= ĠnumberOf Calls Consum ers Ġ Ġ. add ( consumer ) Ċ ĉ fun Ġadd Consumer For Processing Events ( consumer : ĠKafka Entity Consumer < Processing Event >) ĠĠ Ġ= Ġprocessing Event Consum ers . add ( consumer ) Ċ } Ċ