package Ġstreams . kafka Ċ Ċ import Ġorg . apache . kafka . clients . consumer . Offset And Metadata Ċ import Ġorg . apache . kafka . common . Topic Partition Ċ import Ġorg . neo 4 j . logging . Log Ċ import Ġstreams . service . Streams Sink Entity Ċ import Ġstreams . service . errors . Error Service Ċ Ċ class ĠKafka Manual Commit Event Consumer ( private Ġval Ġconfig : ĠKafka Sink Configuration , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġlog : ĠLog , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprivate Ġval Ġdl q Service : ĠError Service ): ĠKafka Auto Commit Event Consumer ( config , Ġlog , Ġdl q Service ) Ġ{ ĊĊĠĠĠ Ġprivate Ġfun Ġcommit Data ( commit : ĠBoolean , Ġtopic Map : ĠMap < Topic Partition , ĠOffset And Metadata >) Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( commit ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer . commit Sync ( topic Map ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġread ( action : Ġ( String , ĠList < Streams Sink Entity >) Ġ-> ĠUnit ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġtopic Map Ġ= Ġread Simple ( action ) ĊĠĠĠĠĠĠĠ Ġcommit Data ( true , Ġtopic Map ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġread ( topic Config : ĠMap < String , ĠAny >, Ġaction : Ġ( String , ĠList < Streams Sink Entity >) Ġ-> ĠUnit ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġkafka Topic Config Ġ= ĠKafka Topic Config . from Map ( topic Config ) ĊĠĠĠĠĠĠĠ Ġval Ġtopic Map Ġ= Ġif Ġ( kafka Topic Config . topic Partitions Map . isEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġread Simple ( action ) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġread From Partition ( kafka Topic Config , Ġaction ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġcommit Data ( kafka Topic Config . commit , Ġtopic Map ) ĊĠĠĠ Ġ} Ċ }