package Ġco . nil in . op ex . wallet . ports . kafka . listener . consumer ĊĊ Ċ import Ġco . nil in . op ex . wallet . ports . kafka . listener . model . User Created Event Ċ import Ġco . nil in . op ex . wallet . ports . kafka . listener . spi . User Created EventListener Ċ import Ġorg . apache . kafka . clients . consumer . Consumer Record Ċ import Ġorg . springframework . kafka . listener . Message Listener Ċ import Ġorg . springframework . stereotype . Component Ċ Ċ @ Component Ċ class ĠUser Created Kafka Listener Ġ: ĠMessage Listener < String , ĠUser Created Event > Ġ{ ĊĠĠĠ Ġval Ġevent Listeners Ġ= Ġarray ListOf < User Created EventListener >() ĊĊĠĠĠ Ġoverride Ġfun Ġon Message ( data : ĠConsumer Record < String , ĠUser Created Event >) Ġ{ ĊĠĠĠĠĠĠĠ Ġevent Listeners . forEach Ġ{ Ġtl Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtl . on Event ( data . value (), Ġdata . partition (), Ġdata . offset (), Ġdata . timestamp ()) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġadd EventListener ( tl : ĠUser Created EventListener ) Ġ{ ĊĠĠĠĠĠĠĠ Ġevent Listeners . add ( tl ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġremove EventListener ( tl : ĠUser Created EventListener ) Ġ{ ĊĠĠĠĠĠĠĠ Ġevent Listeners . remove If Ġ{ Ġitem Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġitem . id () Ġ== Ġtl . id () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ }