package Ġcom . proc ure ment . fel icia . infrastructure . kafka Ċ Ċ import Ġcom . proc ure ment . fel icia . application . network . model . Socket Ċ import Ġorg . apache . kafka . clients . Common Client Configs Ċ import Ġorg . apache . kafka . clients . consumer . Consumer Config Ċ import Ġorg . apache . kafka . clients . consumer . Kafka Consumer Ċ import Ġorg . apache . kafka . common . config . S asl Configs Ċ import Ġorg . apache . kafka . common . serialization . String Deserializer Ċ import Ġjava . util .* Ċ Ċ fun Ġcreate Kafka Consumer ( hosts : ĠList < Socket >, Ġauth Credentials : ĠKafka Auth Credentials , ĠgroupId : ĠString ): ĠKafka Consumer < String , ĠString > Ġ{ ĊĊĠĠĠ Ġval Ġprops Ġ= ĠProperties () ĊĠĠĠ Ġprops [ Consumer Config . BOOT STR AP _ SERVER S _ CONFIG ] Ġ= Ġhosts . map Ġ{ Ġit . toString () Ġ} ĊĠĠĠ Ġprops [ Consumer Config . GROUP _ ID _ CONFIG ] Ġ= ĠgroupId ĊĠĠĠ Ġprops [ Consumer Config . KEY _ DE SERIAL IZER _ CLASS _ CONFIG ] Ġ= ĠString Deserializer :: class . java ĊĠĠĠ Ġprops [ Consumer Config . VALUE _ DE SERIAL IZER _ CLASS _ CONFIG ] Ġ= ĠString Deserializer :: class . java ĊĠĠĠ Ġprops [ Consumer Config . ENABLE _ AUTO _ COMMIT _ CONFIG ] Ġ= Ġfalse ĊĠĠĠ Ġprops [ Consumer Config . AUTO _ OFFSET _ RESET _ CONFIG ] Ġ= Ġ" latest " ĊĊĠĠĠ Ġwhen Ġ( auth Credentials ) Ġ{ ĊĠĠĠĠĠĠĠ Ġis ĠKafka Auth Credentials . S asl Auth Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprops [ Common Client Configs . SECURITY _ PROTOCOL _ CONFIG ] Ġ= Ġ" SAS L _ PL AINT EXT " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprops [ S asl Configs . SAS L _ ME CHAN IS M ] Ġ= Ġ" PLAIN " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprops [ S asl Configs . SAS L _ JA AS _ CONFIG ] Ġ= Ġauth Credentials . SAS L _ JA AS ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠĠĠ Ġreturn ĠKafka Consumer ( props ) Ċ } Ċ