package Ġfund . cyber . supply Ċ Ċ import Ġfund . cyber . common . kafka . Json Serializer Ċ import Ġfund . cyber . common . kafka . default Producer Config Ċ import Ġfund . cyber . common . with Ċ import Ġfund . cyber . search . configuration . CHAIN _ FAMILY Ċ import Ġfund . cyber . search . configuration . CHAIN _ NAME Ċ import Ġfund . cyber . search . configuration . K AF KA _ BRO K ERS Ċ import Ġfund . cyber . search . configuration . K AF KA _ BRO K ERS _ DEFAULT Ċ import Ġfund . cyber . search . model . chains . Chain Family Ċ import Ġfund . cyber . search . model . chains . Chain Info Ċ import Ġfund . cyber . search . model . events . Pump Event Ċ import Ġfund . cyber . search . model . events . supply Topic Ċ import Ġorg . apache . kafka . clients . admin . Admin Client Ċ import Ġorg . apache . kafka . clients . admin . Admin ClientConfig Ċ import Ġorg . apache . kafka . clients . admin . New Topic Ċ import Ġorg . apache . kafka . clients . producer . Producer Config Ċ import Ġorg . apache . kafka . common . config . Topic Config Ċ import Ġorg . springframework . beans . factory . annotation . Value Ċ import Ġorg . springframework . context . annotation . Bean Ċ import Ġorg . springframework . context . annotation . Component Scan Ċ import Ġorg . springframework . context . annotation . Configuration Ċ import Ġorg . springframework . kafka . core . Default Kafka Producer Factory Ċ import Ġorg . springframework . kafka . core . Kafka Admin Ċ import Ġorg . springframework . kafka . core . Kafka Template Ċ import Ġorg . springframework . kafka . core . Producer Factory Ċ import Ġorg . springframework . kafka . transaction . Kafka Transaction Manager Ċ import Ġjavax . annotation . Post Construct ĊĊ Ċ @ Configuration Ċ @ Component Scan (" fund . cyber . supply ") Ċ class ĠCommon Configuration Ġ{ ĊĊĠĠĠ Ġ@ Value ("\ ${ $ K AF KA _ BRO K ERS :$ K AF KA _ BRO K ERS _ DEFAULT }") ĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġkafka Bro kers : ĠString ĊĊĠĠĠ Ġ@ Value ("\ ${ $ CHAIN _ FAMILY : }") ĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġchain Family : ĠString ĊĊĠĠĠ Ġ@ Value ("\ ${ $ CHAIN _ NAME : }") ĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġchain Name : ĠString ĊĊĊĠĠĠ Ġ@ Bean ĊĠĠĠ Ġfun Ġchain Info () Ġ= ĠChain Info ( ĊĠĠĠĠĠĠĠ ĠChain Family . valueOf ( chain Family ), ĊĠĠĠĠĠĠĠ Ġif Ġ( chain Name . isEmpty ()) Ġchain Family Ġelse Ġchain Name ĊĠĠĠ Ġ) ĊĊĠĠĠ Ġ@ Bean ĊĠĠĠ Ġfun Ġproducer Factory (): ĠProducer Factory < Pump Event , ĠAny > Ġ{ ĊĊĠĠĠĠĠĠĠ Ġval Ġconfig Ġ= Ġdefault Producer Config (). with ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠProducer Config . BOOT STR AP _ SERVER S _ CONFIG Ġto Ġkafka Bro kers ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġreturn ĠDefault Kafka Producer Factory < Pump Event , ĠAny >( config , ĠJsonSerializer (), ĠJsonSerializer ()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. apply Ġ{ Ġset Transaction Id Prefix ( chain Info (). name Ġ+ Ġ"_ SUP PLY ") Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Bean ĊĠĠĠ Ġfun Ġtransaction Manager () Ġ= ĠKafka Transaction Manager ( producer Factory ()) ĊĊĠĠĠ Ġ@ Bean ĊĠĠĠ Ġfun Ġkafka Template () Ġ= ĠKafka Template ( producer Factory ()) ĊĊĊĠĠĠ Ġ@ Bean ĊĠĠĠ Ġfun Ġkafka Admin (): ĠKafka Admin Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġconfigs Ġ= Ġmap Of ( Admin ClientConfig . BOOT STR AP _ SERVER S _ CONFIG Ġto Ġkafka Bro kers ) ĊĠĠĠĠĠĠĠ Ġreturn ĠKafka Admin ( configs ). apply Ġ{ Ġthis . set Fatal If Broker Not Available ( true ) Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Post Construct ĊĠĠĠ Ġfun Ġcreate Topics () Ġ{ ĊĊĠĠĠĠĠĠĠ Ġval Ġkafka Client Ġ= ĠAdmin Client . create ( kafka Admin (). config ) ĊĊĠĠĠĠĠĠĠ Ġval Ġsupply Topic Config Ġ= Ġmap Of ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠTopic Config . RET ENT ION _ BYTES _ CONFIG Ġto Ġ" 5 2 4 2 8 8 0 0 ", ĊĠĠĠĠĠĠĠĠĠĠĠ ĠTopic Config . CLEAN UP _ POLICY _ CONFIG Ġto ĠTopic Config . CLEAN UP _ POLICY _ DELETE ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġval Ġsupply Topic Ġ= ĠNew Topic ( chain Info (). supply Topic , Ġ 1 , Ġ 1 ). configs ( supply Topic Config ) ĊĊĠĠĠĠĠĠĠ Ġkafka Client . create Topics ( listOf ( supply Topic )) ĊĠĠĠĠĠĠĠ Ġkafka Client . close () ĊĠĠĠ Ġ} ĊĊ Ċ }