package Ġno . nav . crm . kafka . activity Ċ Ċ import Ġjava . util . Properties Ċ 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 . Ssl Configs Ċ import Ġorg . apache . kafka . common . serialization . String Deserializer Ċ import Ġorg . apache . kafka . common . serialization . String Serializer Ċ Ċ // ĠApp Ġenvironment Ġdependencies Ċ const Ġval ĠEV _ kafka Topic Ġ= Ġ" K AF KA _ TOPIC " Ċ const Ġval ĠEV _ kafka Client ID Ġ= Ġ" K AF KA _ CLIENT ID " Ċ Ċ // ĠEnvironment Ġdependencies Ġinjected Ġin Ġpod Ġby Ġna is Ġkafka Ġsolution Ċ const Ġval ĠEV _ kafka Bro kers Ġ= Ġ" K AF KA _ BRO K ERS " Ċ const Ġval ĠEV _ kafka Ke yst ore Path Ġ= Ġ" K AF KA _ KEY STORE _ PATH " Ċ const Ġval ĠEV _ kafka Cred store Password Ġ= Ġ" K AF KA _ CRED STORE _ PASSWORD " Ċ const Ġval ĠEV _ kafka Trust store Path Ġ= Ġ" K AF KA _ TR UST STORE _ PATH " Ċ Ċ val Ġtopic Ġ= Ġfetch Env ( EV _ kafka Topic ) Ċ Ċ val Ġkafka Producer Config : ĠProperties Ġ= Ġmap Of < String , ĠAny >( ĊĠĠĠ ĠProducer Config . KEY _ SERIAL IZER _ CLASS _ CONFIG Ġto ĠString Serializer :: class . java , ĊĠĠĠ ĠProducer Config . VALUE _ SERIAL IZER _ CLASS _ CONFIG Ġto ĠString Serializer :: class . java , ĊĠĠĠ ĠProducer Config . BOOT STR AP _ SERVER S _ CONFIG Ġto Ġfetch Env ( EV _ kafka Bro kers ), ĊĠĠĠ ĠProducer Config . CLIENT _ ID _ CONFIG Ġto Ġfetch Env ( EV _ kafka Client ID ), ĊĠĠĠ ĠProducer Config . ACK S _ CONFIG Ġto Ġ" all ", ĊĠĠĠ ĠProducer Config . DE LI VERY _ TIMEOUT _ MS _ CONFIG Ġto Ġ 6 0 0 0 0 , ĊĠĠĠ ĠCommon Client Configs . SECURITY _ PROTOCOL _ CONFIG Ġto Ġ" SSL ", ĊĠĠĠ ĠS sl Configs . SSL _ KEY STORE _ LOCATION _ CONFIG Ġto Ġfetch Env ( EV _ kafka Ke yst ore Path ), ĊĠĠĠ ĠS sl Configs . SSL _ KEY STORE _ PASSWORD _ CONFIG Ġto Ġfetch Env ( EV _ kafka Cred store Password ), ĊĠĠĠ ĠS sl Configs . SSL _ TR UST STORE _ LOCATION _ CONFIG Ġto Ġfetch Env ( EV _ kafka Trust store Path ), ĊĠĠĠ ĠS sl Configs . SSL _ TR UST STORE _ PASSWORD _ CONFIG Ġto Ġfetch Env ( EV _ kafka Cred store Password )) ĊĠĠĠ Ġ. as Properties () Ċ Ċ val Ġkafka Consumer Config : ĠProperties Ġ= Ġmap Of < String , ĠAny >( ĊĠĠĠ ĠConsumer Config . KEY _ DE SERIAL IZER _ CLASS _ CONFIG Ġto ĠString Deserializer :: class . java , ĊĠĠĠ ĠConsumer Config . VALUE _ DE SERIAL IZER _ CLASS _ CONFIG Ġto ĠString Deserializer :: class . java , ĊĠĠĠ ĠConsumer Config . BOOT STR AP _ SERVER S _ CONFIG Ġto Ġfetch Env ( EV _ kafka Bro kers ), ĊĠĠĠ ĠConsumer Config . GROUP _ ID _ CONFIG Ġto Ġfetch Env ( EV _ kafka Client ID ), ĊĠĠĠ ĠConsumer Config . CLIENT _ ID _ CONFIG Ġto Ġfetch Env ( EV _ kafka Client ID ), ĊĠĠĠ ĠConsumer Config . AUTO _ OFFSET _ RESET _ CONFIG Ġto Ġ" ear liest ", ĊĠĠĠ ĠConsumer Config . MAX _ POLL _ RECORD S _ CONFIG Ġto Ġ 2 0 0 , Ġ// ĠMatch Ġmaximum Ġwhen Ġsending Ġto ĠSF ĠREST ĠAPI ĊĠĠĠ ĠConsumer Config . ENABLE _ AUTO _ COMMIT _ CONFIG Ġto Ġ" false ", Ġ// ĠDo Ġnot Ġcommit Ġuntil Ġoperation Ġwith Ġdata Ġis Ġconfirmed ĊĠĠĠ ĠCommon Client Configs . SECURITY _ PROTOCOL _ CONFIG Ġto Ġ" SSL ", ĊĠĠĠ ĠS sl Configs . SSL _ KEY STORE _ LOCATION _ CONFIG Ġto Ġfetch Env ( EV _ kafka Ke yst ore Path ), ĊĠĠĠ ĠS sl Configs . SSL _ KEY STORE _ PASSWORD _ CONFIG Ġto Ġfetch Env ( EV _ kafka Cred store Password ), ĊĠĠĠ ĠS sl Configs . SSL _ TR UST STORE _ LOCATION _ CONFIG Ġto Ġfetch Env ( EV _ kafka Trust store Path ), ĊĠĠĠ ĠS sl Configs . SSL _ TR UST STORE _ PASSWORD _ CONFIG Ġto Ġfetch Env ( EV _ kafka Cred store Password )) ĊĠĠĠ Ġ. as Properties () Ċ Ċ fun ĠMap < String , ĠAny >. as Properties (): ĠProperties Ġ{ ĊĠĠĠ Ġval Ġprops Ġ= ĠProperties () ĊĠĠĠ Ġthis . forEach Ġ{ Ġ( k , Ġv ) Ġ-> Ġprops [ k ] Ġ= Ġv Ġ} ĊĠĠĠ Ġreturn Ġprops Ċ } Ċ Ċ fun Ġfetch Env ( env : ĠString ): ĠString Ġ{ ĊĠĠĠ Ġreturn ĠSystem . getenv ( env ) Ċ } Ċ