package Ġme . ji ay u . k af k akt Ċ Ċ import Ġkotlinx . coroutines . cancel And Join Ċ import Ġkotlinx . coroutines . coroutine Scope Ċ import Ġkotlinx . coroutines . delay Ċ import Ġkotlinx . coroutines . launch Ċ import Ġme . ji ay u . k af k akt . App . Companion . logger Ċ import Ġme . ji ay u . k af k akt . App . Companion . prop Ċ import Ġorg . apache . kafka . clients . consumer . Kafka Consumer Ċ import Ġorg . apache . kafka . clients . producer . Kafka Producer Ċ import Ġorg . apache . kafka . clients . producer . Producer Record Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġjava . util .* Ċ import Ġkotlin . time . Duration . Companion . seconds Ċ import Ġkotlin . time . to Java Duration Ċ Ċ class ĠApp Ġ{ ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġlogger Ġ= ĠLoggerFactory . getLogger ( App :: class . java . name ) !! ĊĊĠĠĠĠĠĠĠ Ġval Ġprop Ġ= Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġprop Ġ= ĠProperties () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprop [" application . id "] Ġ= Ġ" k af k akt " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprop [" group . id "] Ġ= Ġ" k af k akt - 0 0 0 1 " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprop [" bootstrap . servers "] Ġ= Ġ" localhost : 9 0 9 2 " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprop [" key . serializer "] Ġ= Ġ" org . apache . kafka . common . serialization . String Serializer " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprop [" value . serializer "] Ġ= Ġ" org . apache . kafka . common . serialization . String Serializer " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprop [" key . de serializer "] Ġ= Ġ" org . apache . kafka . common . serialization . String Deserializer " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprop [" value . de serializer "] Ġ= Ġ" org . apache . kafka . common . serialization . String Deserializer " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprop ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } ĊĊ Ċ suspend Ġfun Ġmain () Ġ= Ġcoroutine Scope Ġ{ ĊĠĠĠ Ġval Ġtopic Ġ= Ġ" text - topic " ĊĠĠĠ Ġlogger . info (" starting Ġto Ġlisten Ġto Ġ$ topic ") ĊĊĠĠĠ Ġval Ġhandle Ġ= Ġlaunch Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġconsumer Ġ= ĠKafka Consumer < String , ĠString >( prop ()) ĊĠĠĠĠĠĠĠ Ġconsumer . use Ġ{ Ġc Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġc . subscribe ( Collections . singleton (" text - topic ")) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( true ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġrecords Ġ= Ġconsumer . poll ( 1 . 0 . seconds . to Java Duration ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrecords . forEach Ġ{ Ġlogger . info (" key =${ it . key ()}, Ġvalue =${ it . value () }") Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġval Ġproducer Ġ= ĠKafka Producer < String , ĠString >( prop ()) ĊĠĠĠ Ġproducer . use Ġ{ Ġp Ġ-> ĊĠĠĠĠĠĠĠ Ġ( 0 .. 1 0 0 ). forEach Ġ{ Ġp . send ( Producer Record (" text - topic ", Ġ"$ it ", Ġ" value Ġ$ it ")) Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġdelay ( 4 . 0 . seconds ) ĊĠĠĠ Ġlogger . info (" c ancell ing ") ĊĠĠĠ Ġhandle . cancel And Join () Ċ } ĊĊ