package Ġno . nav . dp . ved t aks ly t ter Ċ Ċ import Ġcom . nat pr y ce . kon fig . Configuration Map Ċ import Ġcom . nat pr y ce . kon fig . Configuration Properties Ċ import Ġcom . nat pr y ce . kon fig . Environment Variables Ċ import Ġcom . nat pr y ce . kon fig . Key Ċ import Ġcom . nat pr y ce . kon fig . int Type Ċ import Ġcom . nat pr y ce . kon fig . over riding Ċ import Ġcom . nat pr y ce . kon fig . string Type Ċ import Ġio . con fluent . kafka . serializers . Kafka Av ro Deserializer Config Ċ import Ġno . nav . dag p enger . ktor . auth . ApiKey Verifier Ċ import Ġno . nav . dp . ved t aks ly t ter . avro . Av ro Deserializer Ċ import Ġorg . apache . kafka . clients . Common Client Configs Ċ import Ġorg . apache . kafka . clients . consumer . Consumer Config Ċ import Ġorg . apache . kafka . clients . producer . Producer Config Ċ import Ġorg . apache . kafka . common . config . S asl Configs Ċ 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 . apache . kafka . common . serialization . String Serializer Ċ import Ġjava . io . File Ċ import Ġjava . util . Properties Ċ Ċ private Ġval Ġlocal Properties Ġ= ĠConfiguration Map ( ĊĠĠĠ Ġmap Of ( ĊĠĠĠĠĠĠĠ Ġ" application . profile " Ġto Ġ" LOCAL ", ĊĠĠĠĠĠĠĠ Ġ" application . http Port " Ġto Ġ" 8 0 9 9 ", ĊĠĠĠĠĠĠĠ Ġ" kafka . bootstrap Server " Ġto Ġ" localhost : 9 0 9 2 ", ĊĠĠĠĠĠĠĠ Ġ" kafka . schema Registry Server " Ġto Ġ" http :// localhost : 8 0 8 1 ", ĊĠĠĠĠĠĠĠ Ġ" kafka . topic " Ġto Ġ" priv at - arena - dag pen ge ved tak - fer dig st ilt ", ĊĠĠĠĠĠĠĠ Ġ" username " Ġto Ġ" srv dp - ved tak result at ", ĊĠĠĠĠĠĠĠ Ġ" password " Ġto Ġ" ik ken oe ", ĊĠĠĠĠĠĠĠ Ġ" kafka . groupId " Ġto Ġ" srv dp - ved tak result at - ly t ter - v 1 ", ĊĠĠĠĠĠĠĠ Ġ" re gel . api . url " Ġto Ġ" http :// dp - re gel - api . na is . pre prod . local ", ĊĠĠĠĠĠĠĠ Ġ" oidc . sts . issuer url " Ġto Ġ" http :// localhost ", ĊĠĠĠĠĠĠĠ Ġ" srv dp . ved tak result at . ly t ter . username " Ġto Ġ" srv dp - ved tak result at ", ĊĠĠĠĠĠĠĠ Ġ" srv dp . ved tak result at . ly t ter . password " Ġto Ġ" srv dp - pass ord ", ĊĠĠĠĠĠĠĠ Ġ" auth . re gel api . secret " Ġto Ġ" secret ", ĊĠĠĠĠĠĠĠ Ġ" auth . re gel api . key " Ġto Ġ" key " ĊĠĠĠ Ġ) Ċ ) Ċ Ċ private Ġval Ġdev Properties Ġ= ĠConfiguration Map ( ĊĠĠĠ Ġmap Of ( ĊĠĠĠĠĠĠĠ Ġ" application . profile " Ġto Ġ" DEV ", ĊĠĠĠĠĠĠĠ Ġ" application . http Port " Ġto Ġ" 8 0 9 9 ", ĊĠĠĠĠĠĠĠ Ġ" kafka . bootstrap Server " Ġto Ġ" b 2 7 ap vl 0 0 0 4 5 . pre prod . local : 8 4 4 3 , b 2 7 ap vl 0 0 0 4 6 . pre prod . local : 8 4 4 3 , b 2 7 ap vl 0 0 0 4 7 . pre prod . local : 8 4 4 3 ", ĊĠĠĠĠĠĠĠ Ġ" kafka . schema Registry Server " Ġto Ġ" https :// kafka - schema - registry . na is . pre prod . local ", ĊĠĠĠĠĠĠĠ Ġ" kafka . topic " Ġto Ġ" priv at - arena - dag pen ge ved tak - fer dig st ilt ", ĊĠĠĠĠĠĠĠ Ġ" username " Ġto Ġ" srv dp - ved tak result at ", ĊĠĠĠĠĠĠĠ Ġ" password " Ġto Ġ" ik ken oe ", ĊĠĠĠĠĠĠĠ Ġ" re gel . api . url " Ġto Ġ" http :// dp - re gel - api . default ", ĊĠĠĠĠĠĠĠ Ġ" oidc . sts . issuer url " Ġto Ġ" http :// localhost ", ĊĠĠĠĠĠĠĠ Ġ" srv dp . ved tak result at . ly t ter . username " Ġto Ġ" srv dp - ved tak result at ", ĊĠĠĠĠĠĠĠ Ġ" srv dp . ved tak result at . ly t ter . password " Ġto Ġ" srv dp - pass ord ", ĊĠĠĠĠĠĠĠ Ġ" kafka . groupId " Ġto Ġ" srv dp - ved tak result at - ly t ter - v 1 ", ĊĠĠĠĠĠĠĠ Ġ" auth . re gel api . secret " Ġto Ġ" secret ", ĊĠĠĠĠĠĠĠ Ġ" auth . re gel api . key " Ġto Ġ" key " ĊĠĠĠ Ġ) Ċ ) Ċ Ċ private Ġval Ġprod Properties Ġ= ĠConfiguration Map ( ĊĠĠĠ Ġmap Of ( ĊĠĠĠĠĠĠĠ Ġ" application . profile " Ġto Ġ" PRO D ", ĊĠĠĠĠĠĠĠ Ġ" application . http Port " Ġto Ġ" 8 0 9 9 ", ĊĠĠĠĠĠĠĠ Ġ" kafka . bootstrap Server " Ġto Ġ" a 0 1 ap vl 0 0 1 4 5 . ade o . no : 8 4 4 3 , a 0 1 ap vl 0 0 1 4 6 . ade o . no : 8 4 4 3 , a 0 1 ap vl 0 0 1 4 7 . ade o . no : 8 4 4 3 , a 0 1 ap vl 0 0 1 4 9 . ade o . no : 8 4 4 3 ", ĊĠĠĠĠĠĠĠ Ġ" kafka . schema Registry Server " Ġto Ġ" http :// kafka - schema - registry . tp a : 8 0 8 1 ", ĊĠĠĠĠĠĠĠ Ġ" kafka . topic " Ġto Ġ" priv at - arena - dag pen ge ved tak - fer dig st ilt ", ĊĠĠĠĠĠĠĠ Ġ" username " Ġto Ġ" srv dp - ved tak result at ", ĊĠĠĠĠĠĠĠ Ġ" password " Ġto Ġ" ik ken oe ", ĊĠĠĠĠĠĠĠ Ġ" kafka . groupId " Ġto Ġ" srv dp - ved tak result at - ly t ter - v 1 ", ĊĠĠĠĠĠĠĠ Ġ" re gel . api . url " Ġto Ġ" http :// dp - re gel - api . default ", ĊĠĠĠĠĠĠĠ Ġ" srv dp . ved tak result at . ly t ter . username " Ġto Ġ" srv dp - ved tak result at ", ĊĠĠĠĠĠĠĠ Ġ" srv dp . ved tak result at . ly t ter . password " Ġto Ġ" srv dp - pass ord ", ĊĠĠĠĠĠĠĠ Ġ" auth . re gel api . secret " Ġto Ġ" secret ", ĊĠĠĠĠĠĠĠ Ġ" auth . re gel api . key " Ġto Ġ" key " ĊĠĠĠ Ġ) Ċ ) Ċ Ċ data Ġclass ĠApplication ( ĊĠĠĠ Ġval Ġhttp Port : ĠInt Ġ= Ġconfig ()[ Key (" application . http Port ", Ġint Type )], ĊĠĠĠ Ġval Ġprofile : ĠProfile Ġ= Ġconfig ()[ Key (" application . profile ", Ġstring Type )]. let Ġ{ ĠProfile . valueOf ( it ) Ġ}, ĊĠĠĠ Ġval Ġusername : ĠString Ġ= Ġconfig ()[ Key (" srv dp . ved tak result at . ly t ter . username ", Ġstring Type )], ĊĠĠĠ Ġval Ġpassword : ĠString Ġ= Ġconfig ()[ Key (" srv dp . ved tak result at . ly t ter . password ", Ġstring Type )] Ċ ) Ċ Ċ data Ġclass ĠKafka ( ĊĠĠĠ Ġval Ġbootstrap Server : ĠString Ġ= Ġconfig ()[ Key (" kafka . bootstrap Server ", Ġstring Type )], ĊĠĠĠ Ġval Ġschema Registry Server : ĠString Ġ= Ġconfig ()[ Key (" kafka . schema Registry Server ", Ġstring Type )], ĊĠĠĠ Ġval Ġtopic : ĠString Ġ= Ġconfig ()[ Key (" kafka . topic ", Ġstring Type )], ĊĠĠĠ Ġval Ġusername : ĠString Ġ= Ġconfig ()[ Key (" username ", Ġstring Type )], ĊĠĠĠ Ġval Ġpassword : ĠString Ġ= Ġconfig ()[ Key (" password ", Ġstring Type )], ĊĠĠĠ Ġval ĠgroupId : ĠString Ġ= Ġconfig ()[ Key (" kafka . groupId ", Ġstring Type )] Ċ ) Ġ{ ĊĊĠĠĠ Ġfun Ġto Consumer Props (): ĠProperties Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠProperties (). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Consumer Config . BOOT STR AP _ SERVER S _ CONFIG , Ġbootstrap Server ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Kafka Av ro Deserializer Config . SCHEMA _ REGISTRY _ URL _ CONFIG , Ġschema Registry Server ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Consumer Config . GROUP _ ID _ CONFIG , ĠgroupId ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Consumer Config . KEY _ DE SERIAL IZER _ CLASS _ CONFIG , ĠString Deserializer :: class . java ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Consumer Config . VALUE _ DE SERIAL IZER _ CLASS _ CONFIG , ĠAv ro Deserializer :: class . java ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Consumer Config . AUTO _ OFFSET _ RESET _ CONFIG , Ġ" ear liest ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Consumer Config . ENABLE _ AUTO _ COMMIT _ CONFIG , Ġfalse ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput All ( credentials ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġto A iven Producer Props (): ĠProperties Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġenv : ĠMap < String , ĠString > Ġ= Ġconfig (). list (). reversed (). fold ( empty Map ()) Ġ{ Ġmap , Ġpair Ġ-> Ġmap Ġ+ Ġpair . second Ġ} ĊĠĠĠĠĠĠĠ Ġreturn ĠProperties (). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Common Client Configs . BOOT STR AP _ SERVER S _ CONFIG , Ġenv . getValue (" 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 . getValue (" K AF KA _ TR UST STORE _ PATH ")) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Ssl Configs . SSL _ TR UST STORE _ PASSWORD _ CONFIG , Ġenv . getValue (" K AF KA _ CRED STORE _ PASSWORD ")) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Ssl Configs . SSL _ KEY STORE _ LOCATION _ CONFIG , Ġenv . getValue (" K AF KA _ KEY STORE _ PATH ")) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Ssl Configs . SSL _ KEY STORE _ PASSWORD _ CONFIG , Ġenv . getValue (" K AF KA _ CRED STORE _ PASSWORD ")) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Producer Config . ACK S _ CONFIG , Ġ" 1 ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Producer Config . LING ER _ MS _ CONFIG , Ġ" 0 ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Producer Config . MAX _ IN _ FL IGHT _ REQUEST S _ PER _ CONNECTION , Ġ" 1 ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Producer Config . KEY _ SERIAL IZER _ CLASS _ CONFIG , ĠString Serializer :: class . java ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Producer Config . VALUE _ SERIAL IZER _ CLASS _ CONFIG , ĠString Serializer :: class . java ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Producer Config . CLIENT _ ID _ CONFIG , Ġ" dp - ved tak result at - ly t ter ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġcredentials (): ĠProperties Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠProperties (). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ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 ="$ username " Ġpassword ="$ password "; """ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠSystem . getenv (" NAV _ TR UST STORE _ PATH ") ?. let Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ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 ")) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ Ċ enum Ġclass ĠProfile Ġ{ ĊĠĠĠ ĠLOCAL , ĠDEV , ĠPRO D Ċ } Ċ Ċ data Ġclass ĠConfiguration ( ĊĠĠĠ Ġval Ġapplication : ĠApplication Ġ= ĠApplication (), ĊĠĠĠ Ġval Ġkafka : ĠKafka Ġ= ĠKafka (), ĊĠĠĠ Ġval Ġre gel Api Url : ĠString Ġ= Ġconfig ()[ Key (" re gel . api . url ", Ġstring Type )], ĊĠĠĠ Ġval Ġauth : ĠAuth Ġ= ĠAuth () Ċ ) Ċ Ċ class ĠAuth ( ĊĠĠĠ Ġre gel Api Secret : ĠString Ġ= Ġconfig ()[ Key (" auth . re gel api . secret ", Ġstring Type )], ĊĠĠĠ Ġre gel ApiKey Plain : ĠString Ġ= Ġconfig ()[ Key (" auth . re gel api . key ", Ġstring Type )] Ċ ) Ġ{ ĊĠĠĠ Ġval Ġre gel ApiKey Ġ= ĠApi Key Verifier ( re gel Api Secret ). generate ( re gel ApiKey Plain ) Ċ } Ċ Ċ fun Ġget Env Or Prop ( prop Name : ĠString ): ĠString ? Ġ{ ĊĠĠĠ Ġreturn ĠSystem . getenv ( prop Name ) Ġ?: ĠSystem . getProperty ( prop Name ) Ċ } Ċ Ċ private Ġfun Ġconfig () Ġ= Ġwhen Ġ( get Env Or Prop (" NA IS _ CLUSTER _ NAME ")) Ġ{ ĊĠĠĠ Ġ" dev - f ss " Ġ-> ĠConfiguration Properties . system Properties () Ġoverriding ĠEnvironment Variables Ġoverriding Ġdev Properties ĊĠĠĠ Ġ" prod - f ss " Ġ-> ĠConfiguration Properties . system Properties () Ġoverriding ĠEnvironment Variables Ġoverriding Ġprod Properties ĊĠĠĠ Ġelse Ġ-> ĠConfiguration Properties . system Properties () Ġoverriding ĠEnvironment Variables Ġoverriding Ġlocal Properties Ċ } Ċ