package Ġcom . github . wan abe . kafka _ streams _ kotlin _ study Ċ Ċ import Ġorg . apache . kafka . common . serialization . Ser de Ċ import Ġorg . apache . kafka . streams . Kafka Streams Ċ import Ġorg . apache . kafka . streams . Streams Builder Ċ import Ġorg . apache . kafka . streams . k stream . Cons umed Ċ import Ġorg . apache . kafka . streams . k stream . K Stream Ċ import Ġjava . util . Properties Ċ Ċ class ĠStream < K , ĠV >( bro kers : ĠString , Ġvar Ġtopic : ĠString , Ġkey Type : ĠSer de < K >, Ġvalue Type : ĠSer de < V >, Ġf : Ġ( K Stream < K , ĠV >) Ġ-> ĠUnit ) Ġ{ ĊĠĠĠ Ġvar Ġprops Ġ= ĠProperties () ĊĠĠĠ Ġvar Ġstreams Builder Ġ= ĠStreams Builder () ĊĠĠĠ Ġvar Ġkafka Streams : ĠKafka Streams ĊĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠ Ġprops [" bootstrap . servers "] Ġ= Ġbro kers ĊĠĠĠĠĠĠĠ Ġprops [" application . id "] Ġ= Ġ" kafka - streams - sandbox -$ topic " ĊĠĠĠĠĠĠĠ ĠApp . logger . debug (" kafka - streams - sandbox -$ topic Ġ$ streams Builder ") ĊĠĠĠĠĠĠĠ Ġval Ġfirst Stream : ĠK Stream < K , ĠV > Ġ= Ġstreams Builder ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. stream < K , ĠV >( topic , ĠCon su med . with ( key Type , Ġvalue Type )) ĊĠĠĠĠĠĠĠ Ġf ( first Stream ) ĊĠĠĠĠĠĠĠ Ġval Ġtopology Ġ= Ġstreams Builder . build () ĊĠĠĠĠĠĠĠ Ġkafka Streams Ġ= ĠKafka Streams ( topology , Ġprops ) ĊĠĠĠ Ġ} ĊĠĠĠ Ġfun Ġstart ( f : Ġ() Ġ-> ĠUnit ) Ġ{ ĊĠĠĠĠĠĠĠ ĠApp . logger . debug (" start Ġ$ topic ") ĊĠĠĠĠĠĠĠ Ġkafka Streams . start () ĊĠĠĠĠĠĠĠ Ġf () ĊĠĠĠĠĠĠĠ Ġkafka Streams . close () ĊĠĠĠĠĠĠĠ ĠApp . logger . debug (" end Ġ$ topic ") ĊĠĠĠ Ġ} Ċ } Ċ