package Ġru . dat ana . smart . common . ktor . kafka Ċ Ċ import Ġorg . apache . kafka . clients . consumer . Consumer Group Metadata Ċ import Ġorg . apache . kafka . clients . consumer . Consumer Record Ċ import Ġorg . apache . kafka . clients . consumer . Offset And Metadata Ċ import Ġorg . apache . kafka . clients . producer . Callback Ċ import Ġorg . apache . kafka . clients . producer . Producer Ċ import Ġorg . apache . kafka . clients . producer . Producer Record Ċ import Ġorg . apache . kafka . clients . producer . Record Metadata Ċ import Ġorg . apache . kafka . common . Metric Ċ import Ġorg . apache . kafka . common . Metric Name Ċ import Ġorg . apache . kafka . common . Partition Info Ċ import Ġorg . apache . kafka . common . Topic Partition Ċ import Ġorg . apache . kafka . common . record . Timestamp Type Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġjava . time . Duration Ċ import Ġjava . time . Instant Ċ import Ġjava . util . concurrent . Executors Ċ import Ġjava . util . concurrent . Future Ċ import Ġjava . util . concurrent . locks . Re entrant ReadWrite Lock Ċ import Ġkotlin . concurrent . read Ċ import Ġkotlin . concurrent . write Ċ Ċ class ĠTest Producer < K , ĠV >() Ġ: ĠProducer < K , ĠV > Ġ{ ĊĊĠĠĠ Ġprivate Ġval Ġlogger Ġ= ĠLoggerFactory . getLogger ( this :: class . java ) ĊĊĠĠĠ Ġprivate Ġval Ġexecutor Ġ= ĠExec utors . new Single Thread Executor () ĊĠĠĠ Ġprivate Ġval Ġrecords : ĠMutable List < Consumer Record < K , V >> Ġ= Ġmutable ListOf () ĊĠĠĠ Ġprivate Ġval Ġlock Ġ= ĠRe entrant ReadWrite Lock () ĊĊĠĠĠ Ġfun Ġclean () Ġ= Ġlock . write Ġ{ Ġrecords . clear () Ġ} ĊĠĠĠ Ġfun Ġget Sent () Ġ= Ġlock . read Ġ{ Ġrecords . toList () Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġclose () Ġ{} ĊĊĠĠĠ Ġoverride Ġfun Ġclose ( timeout : ĠDuration ?) Ġ{} ĊĊĠĠĠ Ġoverride Ġfun Ġinit Transactions () Ġ{} ĊĊĠĠĠ Ġoverride Ġfun Ġbegin Transaction () Ġ{} ĊĊĠĠĠ Ġoverride Ġfun Ġsend Offsets To Transaction ( ĊĠĠĠĠĠĠĠ Ġoffsets : ĠMutable Map < Topic Partition , ĠOffset And Metadata > ?, ĊĠĠĠĠĠĠĠ Ġconsumer GroupId : ĠString ? ĊĠĠĠ Ġ) Ġ{ ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġsend Offsets To Transaction ( ĊĠĠĠĠĠĠĠ Ġoffsets : ĠMutable Map < Topic Partition , ĠOffset And Metadata > ?, ĊĠĠĠĠĠĠĠ Ġgroup Metadata : ĠConsumer Group Metadata ? ĊĠĠĠ Ġ) Ġ{ ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġcommit Transaction () Ġ{} ĊĊĠĠĠ Ġoverride Ġfun Ġabort Transaction () Ġ{} ĊĊĠĠĠ Ġoverride Ġfun Ġsend ( record : ĠProducer Record < K , ĠV >): ĠFuture < Record Metadata > Ġ{ ĊĠĠĠĠĠĠĠ Ġlock . write Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrecords . add ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠConsumer Record < K , V >( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrecord . topic (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrecord . partition () Ġ?: Ġ 0 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ 0 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrecord . timestamp () Ġ?: ĠInstant . now (). to Epoch Milli (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠTimestamp Type . CREATE _ TIME , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ 0 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ 0 , 0 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrecord . key (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrecord . value () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġlogger . debug (" record Ġis Ġrecorded ") ĊĠĠĠĠĠĠĠ Ġreturn Ġexecutor . submit Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠRecord Metadata ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠTopic Partition ( record . topic (), Ġ 0 ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ 0 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ 0 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠInstant . now (). to Epoch Milli (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ 0 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ 0 , Ġ 0 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} Ġas ĠFuture < Record Metadata > ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġsend ( record : ĠProducer Record < K , ĠV >, Ġcallback : ĠCallback ?): ĠFuture < Record Metadata > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġsend ( record ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġflush () Ġ{} ĊĊĠĠĠ Ġoverride Ġfun Ġpartitions For ( topic : ĠString ?): ĠMutable List < Partition Info > Ġ= Ġmutable ListOf () ĊĊĠĠĠ Ġoverride Ġfun Ġmetrics (): ĠMutable Map < Metric Name , Ġout ĠMetric > Ġ= Ġmutable Map Of () Ċ } Ċ