package Ġli . do er f . micro store . services Ċ Ċ import Ġli . do er f . micro store . utils . getLogger Ċ import Ġorg . apache . kafka . clients . producer . Producer Record Ċ import Ġorg . apache . kafka . common . header . internals . Record Header Ċ import Ġorg . springframework . beans . factory . annotation . Autowired Ċ import Ġorg . springframework . kafka . core . Kafka Template Ċ import Ġorg . springframework . kafka . support . Kafka Headers Ċ import Ġorg . springframework . stereotype . Service Ċ import Ġorg . springframework . util . concurrent . List enable Future Callback Ċ Ċ @ Service Ċ open Ġclass ĠKafka Service Ġ@ Autowired Ġconstructor ( private Ġval Ġkafka Template : ĠKafka Template < String , ĠAny >) Ġ{ ĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġ@ Suppress (" JAVA _ CLASS _ ON _ COMP AN ION ") ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġlog Ġ= Ġget Logger ( java Class ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġopen Ġfun Ġsend Event ( topic : ĠString , Ġevent : ĠAny , Ġcorrelation Id : ĠString ) Ġ{ ĊĠĠĠĠĠĠĠ Ġlog . debug (" sending Ġevent Ġ$ event Ġto Ġtopic Ġ$ topic ") ĊĠĠĠĠĠĠĠ Ġval Ġrecord Ġ= ĠProducer Record < String , ĠAny >( topic , Ġevent ) ĊĠĠĠĠĠĠĠ Ġrecord . headers (). add ( Record Header ( Kafka Headers . COR REL ATION _ ID , Ġcorrelation Id . toByteArray ())) ĊĊĠĠĠĠĠĠĠ Ġval Ġsend Future Ġ= Ġkafka Template . send ( record ) ĊĠĠĠĠĠĠĠ Ġsend Future . add Callback ( object : ĠList enable Future Callback < Any > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun ĠonSuccess ( result : ĠAny ?) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . debug ("$ topic Ġ-> ĠonSuccess ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Failure ( ex : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . error (" topic Ġ-> Ġon Failure ", Ġex ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġopen Ġfun Ġsend Event ( topic : ĠString , Ġkey : ĠString , Ġevent : ĠAny , Ġcorrelation Id : ĠString ) Ġ{ ĊĠĠĠĠĠĠĠ Ġlog . debug (" sending Ġevent Ġ$ event Ġto Ġtopic Ġ$ topic ") ĊĠĠĠĠĠĠĠ Ġval Ġrecord Ġ= ĠProducer Record < String , ĠAny >( topic , Ġkey , Ġevent ) ĊĠĠĠĠĠĠĠ Ġrecord . headers (). add ( Record Header ( Kafka Headers . COR REL ATION _ ID , Ġcorrelation Id . toByteArray ())) ĊĊĠĠĠĠĠĠĠ Ġval Ġsend Future Ġ= Ġkafka Template . send ( record ) ĊĠĠĠĠĠĠĠ Ġsend Future . add Callback ( object : ĠList enable Future Callback < Any > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun ĠonSuccess ( result : ĠAny ?) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . debug ("$ topic Ġ-> ĠonSuccess ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġoverride Ġfun Ġon Failure ( ex : ĠThrowable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . error (" topic Ġ-> Ġon Failure ", Ġex ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠ Ġ} Ċ }