package Ġfund . cyber . pump . kafka Ċ Ċ import Ġfund . cyber . node . common . await All Ċ import Ġfund . cyber . pump . Block Bundle Ċ import Ġfund . cyber . pump . Storage Action Ċ import Ġfund . cyber . pump . Storage Action Source Factory Ċ import Ġorg . apache . kafka . clients . producer . Kafka Producer Ċ import Ġorg . apache . kafka . clients . producer . Producer Record ĊĊ Ċ class ĠKafka Storage Action ( ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġproducer : ĠKafka Producer < Any , ĠAny >, ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġaction Template : ĠKafka Storage Action Template Ċ ) Ġ: ĠStorage Action Ġ{ ĊĊĠĠĠ Ġoverride Ġfun Ġstore () Ġ{ ĊĠĠĠĠĠĠĠ Ġproducer . beginTransaction () ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġaction Template . store Records . map Ġ{ Ġrecord Ġ-> Ġproducer . send ( record ) Ġ}. await All () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġproducer . commit Transaction () ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġproducer . abort Transaction () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠRuntimeException ( e ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġremove () Ġ{ ĊĠĠĠĠĠĠĠ Ġproducer . beginTransaction () ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġaction Template . remove Records . map Ġ{ Ġrecord Ġ-> Ġproducer . send ( record ) Ġ}. await All () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġproducer . commit Transaction () ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġproducer . abort Transaction () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠRuntimeException ( e ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ Ċ interface ĠKafka Storage Action Template Factory < in ĠT Ġ: ĠBlock Bundle > Ġ: ĠStorage Action Source Factory Ġ{ ĊĠĠĠ Ġfun Ġconstruct Action Template ( bundle : ĠT ): ĠKafka Storage Action Template Ċ } ĊĊ Ċ @ Suppress (" UN CHECK ED _ CAST ") Ċ class ĠKafka Storage Action Template ( ĊĠĠĠĠĠĠĠ Ġstore Records : ĠList < Producer Record < *, Ġ* >>, ĊĠĠĠĠĠĠĠ Ġremove Records : ĠList < Producer Record < *, Ġ* >> Ċ ) Ġ{ ĊĠĠĠ Ġval Ġstore Records : ĠList < Producer Record < Any , ĠAny >> Ġ= Ġstore Records Ġas ĠList < Producer Record < Any , ĠAny >> ĊĠĠĠ Ġval Ġremove Records : ĠList < Producer Record < Any , ĠAny >> Ġ= Ġremove Records Ġas ĠList < Producer Record < Any , ĠAny >> Ċ }