/* Ċ Ġ* ĠCopyright Ġ 2 0 1 9 Ġthe Ġoriginal Ġauthor Ġor Ġauthors . Ċ Ġ* Ċ Ġ* ĠLicensed Ġunder Ġthe ĠApache ĠLicense , ĠVersion Ġ 2 . 0 Ġ( the Ġ" License "); Ċ Ġ* Ġyou Ġmay Ġnot Ġuse Ġthis Ġfile Ġexcept Ġin Ġcompliance Ġwith Ġthe ĠLicense . Ċ Ġ* ĠYou Ġmay Ġobtain Ġa Ġcopy Ġof Ġthe ĠLicense Ġat Ċ Ġ* Ċ Ġ* ĠĠĠĠĠ Ġhttps :// www . apache . org / licenses / LICENSE - 2 . 0 Ċ Ġ* Ċ Ġ* ĠUnless Ġrequired Ġby Ġapplicable Ġlaw Ġor Ġagreed Ġto Ġin Ġwriting , Ġsoftware Ċ Ġ* Ġdistributed Ġunder Ġthe ĠLicense Ġis Ġdistributed Ġon Ġan Ġ" AS ĠIS " ĠBASIS , Ċ Ġ* ĠWITHOUT ĠWARRANTIES ĠOR ĠCONDITIONS ĠOF ĠANY ĠKIND , Ġeither Ġexpress Ġor Ġimplied . Ċ Ġ* ĠSee Ġthe ĠLicense Ġfor Ġthe Ġspecific Ġlanguage Ġgoverning Ġpermissions Ġand Ċ Ġ* Ġlimitations Ġunder Ġthe ĠLicense . Ċ Ġ*/ Ċ Ċ package Ġcom . example . kotlin Ċ Ċ import Ġjava . util . concurrent . Concurrent HashMap Ċ Ċ import Ġorg . apache . kafka . clients . admin . New Topic Ċ import Ġorg . apache . kafka . clients . consumer . Consumer Ċ import Ġorg . apache . kafka . common . Topic Partition Ċ import Ġorg . slf 4 j . LoggerFactory Ċ Ċ import Ġorg . springframework . boot . Spring Application Ċ import Ġorg . springframework . boot . autoconfigure . SpringBoot Application Ċ import Ġorg . springframework . context . annotation . Bean Ċ import Ġorg . springframework . kafka . annotation . Kafka Listener Ċ import Ġorg . springframework . kafka . config . Concurrent Kafka Listener Container Factory Ċ import Ġorg . springframework . kafka . config . Topic Builder Ċ import Ġorg . springframework . kafka . listener . Consumer Aware Re balance Listener Ċ import Ġorg . springframework . kafka . support . Kafka Headers Ċ import Ġorg . springframework . messaging . handler . annotation . Header Ċ import Ġorg . springframework . messaging . handler . annotation . Payload Ċ import Ġorg . springframework . stereotype . Component Ċ Ċ @ SpringBoot Application Ċ open Ġclass ĠRead C urrencies Application Ġ{ ĊĊĠĠĠ Ġprivate Ġval Ġc urrencies Ġ= ĠConcurrent HashMap < String , ĠDouble >() ĊĊĠĠĠ Ġ@ Kafka Listener ( id Ġ= Ġ" currency 1 ", Ġtopics Ġ= Ġ[" currency "]) ĊĠĠĠ Ġfun Ġlisten (@ Payload ( required Ġ= Ġfalse ) Ġrate : ĠDouble ?, ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ@ Header ( Kafka Headers . RECE IVED _ MESSAGE _ KEY ) Ġkey : ĠString ) Ġ{ ĊĊĠĠĠĠĠĠĠ Ġif Ġ( rate Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthis . currencies . remove ( key ) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthis . currencies [ key ] Ġ= Ġrate ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġlogger . info (" C urrencies Ġnow : Ġ" Ġ+ Ġthis . currencies ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Bean ĊĠĠĠ Ġopen Ġfun Ġtopic (): ĠNew Topic Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠTopic Builder . name (" currency ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. compact () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. partitions ( 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. replicas ( 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. build () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġlogger Ġ= ĠLoggerFactory . getLogger ( Read C urrencies Application :: class . java ) ĊĊĠĠĠĠĠĠĠ Ġ@ Jvm Static ĊĠĠĠĠĠĠĠ Ġfun Ġmain ( args : ĠArray < String >) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠSpring Application . run ( Read C urrencies Application :: class . java , Ġ* args ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ Ċ } Ċ Ċ @ Component Ċ internal Ġclass ĠFactory Configurer ( factory : ĠConcurrent Kafka Listener Container Factory < *, Ġ* >) Ġ{ ĊĊĠĠĠ Ġinit Ġ{ ĊĠĠĠĠĠĠĠ Ġfactory . container Properties . consumer Re balance Listener Ġ= Ġobject Ġ: ĠConsumer Aware Re balance Listener Ġ{ ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Partitions Assigned ( consumer : ĠConsumer < *, Ġ*> ?, Ġpartitions : ĠCollection < Topic Partition > ?) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer !!. seek To Beginning ( partitions ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ Ċ } Ċ