package Ġno . nav . person br uk er . dit tn av . var sel best iller . config Ċ Ċ import Ġio . con fluent . kafka . serializers . Kafka Av ro Deserializer Config Ċ import Ġio . con fluent . kafka . serializers . Kafka Av ro Serializer Ċ import Ġio . con fluent . kafka . serializers . Kafka Av ro Serializer Config Ċ import Ġio . netty . util . Net Util . get Hostname Ċ import Ġno . nav . person br uk er . dit tn av . var sel best iller . common . kafka . serializer . Sw allow Serialization Errors 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 . serialization . String Serializer Ċ import Ġjava . net . Inet SocketAddress Ċ import Ġjava . util .* Ċ Ċ object ĠKafka Ġ{ ĊĊĠĠĠ Ġprivate Ġconst Ġval Ġtransaction Id Name Ġ= Ġ" dit tn av - var sel best iller - transaction " ĊĊĠĠĠ Ġfun Ġconsumer Props ( env : ĠEnvironment , Ġevent type To Consume : ĠEvent type ): ĠProperties Ġ{ ĊĠĠĠĠĠĠĠ Ġval ĠgroupId And EventType Ġ= Ġbuild GroupId In cluding EventType ( env , Ġevent type To Consume ) ĊĠĠĠĠĠĠĠ Ġreturn ĠProperties (). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Consumer Config . BOOT STR AP _ SERVER S _ CONFIG , Ġenv . a iven Bro kers ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Kafka Av ro Deserializer Config . SCHEMA _ REGISTRY _ URL _ CONFIG , Ġenv . a iven Schema Registry ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Consumer Config . GROUP _ ID _ CONFIG , ĠgroupId And EventType ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Consumer Config . CLIENT _ ID _ CONFIG , ĠgroupId And EventType Ġ+ Ġget Hostname ( Inet SocketAddress ( 0 ))) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Consumer Config . AUTO _ OFFSET _ RESET _ CONFIG , Ġ" ear liest ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Consumer Config . ENABLE _ AUTO _ COMMIT _ CONFIG , Ġfalse ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Consumer Config . KEY _ DE SERIAL IZER _ CLASS _ CONFIG , ĠSw allow Serialization Errors Av ro Deserializer :: class . java ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Consumer Config . VALUE _ DE SERIAL IZER _ CLASS _ CONFIG , ĠSw allow Serialization Errors Av ro Deserializer :: class . java ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Kafka Av ro Deserializer Config . SPEC IFIC _ AV RO _ READ ER _ CONFIG , Ġtrue ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( env . security Config . enabled ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput All ( credential Props A iven ( env . security Config . variables !! )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcredential Props A iven ( security Vars : ĠSecurity Vars ): ĠProperties Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠProperties (). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Kafka Av ro Serializer Config . USER _ INFO _ CONFIG , Ġ"${ security Vars . a iven Schema Registry User }:${ security Vars . a iven Schema Registry Password }") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Kafka Av ro Serializer Config . BASIC _ AUTH _ CRED ENTIAL S _ SOURCE , Ġ" USER _ INFO ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( S asl Configs . SAS L _ ME CHAN IS M , Ġ" PLAIN ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Common Client Configs . SECURITY _ PROTOCOL _ CONFIG , Ġ" SSL ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ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 , Ġsecurity Vars . a iven Trust store Path ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Ssl Configs . SSL _ TR UST STORE _ PASSWORD _ CONFIG , Ġsecurity Vars . a iven Cred store Password ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Ssl Configs . SSL _ KEY STORE _ LOCATION _ CONFIG , Ġsecurity Vars . a iven Ke yst ore Path ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Ssl Configs . SSL _ KEY STORE _ PASSWORD _ CONFIG , Ġsecurity Vars . a iven Cred store Password ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Ssl Configs . SSL _ KEY _ PASSWORD _ CONFIG , Ġsecurity Vars . a iven Cred store Password ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Ssl Configs . SSL _ ENDPOINT _ IDENT IFICATION _ ALGORITHM _ CONFIG , Ġ"") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġproducer Props ( env : ĠEnvironment , Ġtype : ĠEvent type ): ĠProperties Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠProperties (). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Producer Config . BOOT STR AP _ SERVER S _ CONFIG , Ġenv . a iven Bro kers ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Kafka Av ro Serializer Config . SCHEMA _ REGISTRY _ URL _ CONFIG , Ġenv . a iven Schema Registry ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Producer Config . CLIENT _ ID _ CONFIG , Ġenv . groupId Ġ+ Ġget Hostname ( Inet SocketAddress ( 0 ))) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Producer Config . KEY _ SERIAL IZER _ CLASS _ CONFIG , ĠString Serializer :: class . java ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Producer Config . VALUE _ SERIAL IZER _ CLASS _ CONFIG , ĠKafka Av ro Serializer :: class . java ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Producer Config . TRANSACTION AL _ ID _ CONFIG , Ġbuild Transaction Id Name ( type )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Producer Config . MAX _ BLOCK _ MS _ CONFIG , Ġ 4 0 0 0 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Producer Config . ACK S _ CONFIG , Ġ" all ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Producer Config . ENABLE _ ID EMP OT ENCE _ CONFIG , Ġ" true ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( env . security Config . enabled ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput All ( credential Props A iven ( env . security Config . variables !! )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġbuild GroupId In cluding EventType ( env : ĠEnvironment , Ġevent type To Consume : ĠEvent type ) Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠ Ġenv . groupId Ġ+ Ġevent type To Consume . event type ĊĊĠĠĠ Ġprivate Ġfun Ġbuild Transaction Id Name ( event type : ĠEvent type ) Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ"$ transaction Id Name -${ event type . event type }" Ċ Ċ } Ċ