package Ġno . nav . med lems kap . inst . ly t ter Ċ Ċ import Ġno . nav . med lems kap . inst . ly t ter . config . Kafka Config Ċ import Ġorg . apache . kafka . clients . consumer . Consumer Record Ċ import Ġorg . apache . kafka . clients . consumer . Kafka Consumer Ċ import Ġjava . time . Duration Ċ Ċ fun Ġmain ( args : ĠArray < String >) Ġ{ ĊĠĠĠ Ġval Ġconfig : Kafka Config Ġ= ĠKafka Config ( System . getenv ()) ĊĠĠĠ Ġval Ġconsumer : ĠKafka Consumer < String , ĠString > Ġ= Ġconfig . create Consumer () ĊĠĠĠ Ġval Ġmed lems kap V ur der t Topic Ġ= Ġ" med lems kap . med lems kap - v ur der t " ĊĊĠĠĠ Ġconsumer . subscribe ( listOf ( med lems kap V ur der t Topic )) ĊĠĠĠ Ġconsumer . seek To Beginning ( consumer . assignment ()) ĊĊĠĠĠ Ġval Ġpoll Timeout Ġ= ĠDuration . of Seconds ( 2 ) ĊĠĠĠ Ġwhile Ġ( true ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġrecords Ġ= Ġconsumer . poll ( poll Timeout ) ĊĊĠĠĠĠĠĠĠ Ġif Ġ(! records . isEmpty ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrecords . map Ġ{ Ġtransform ( it ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrecords . forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġhandle Message ( it ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer . commit Sync () ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprintln (" ingen Ġme ld inger Ġfun net ĠpÃ¥ Ġk Ã¸ ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ Ċ fun Ġtransform ( it : ĠConsumer Record < String , ĠString > ?) Ġ{ Ċ Ċ } Ċ Ċ fun Ġhandle Message ( it : ĠConsumer Record < String , ĠString > ?) Ġ{ ĊĠĠĠ Ġval Ġtopic Ġ= Ġit ?. topic () ĊĠĠĠ Ġval Ġpartition Ġ= Ġit ?. partition () ĊĠĠĠ Ġval Ġoffset Ġ= Ġit ?. offset () ĊĊĠĠĠ Ġprintln (" topic : Ġ$ topic Ġ, Ġpartition : Ġ$ partition , Ġoffset : Ġ$ offset ") Ċ } Ċ