package Ġstreams . kafka . connect . sink Ċ Ċ import Ġcom . github . jc ust en border . kafka . connect . utils . Version Util Ċ import Ġorg . apache . kafka . connect . sink . Sink Record Ċ import Ġorg . apache . kafka . connect . sink . Sink Task Ċ import Ġorg . slf 4 j . Logger Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġstreams . extensions . as Properties Ċ import Ġstreams . service . errors . Error Data Ċ import Ġstreams . service . errors . Error Service Ċ import Ġstreams . service . errors . Kafka Error Service Ċ import Ġstreams . utils . Streams Utils ĊĊ Ċ class ĠNeo 4 j Sink Task Ġ: ĠS ink Task () Ġ{ ĊĠĠĠ Ġprivate Ġval Ġlog : ĠLogger Ġ= ĠLoggerFactory . getLogger ( Neo 4 j Sink Task :: class . java ) ĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġconfig : ĠNeo 4 j Sink Connector Config ĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġne o 4 j Sink Service : ĠNeo 4 j Sink Service ĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġerror Service : ĠError Service ĊĊĠĠĠ Ġoverride Ġfun Ġversion (): ĠString Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠVersion Util . version ( this . java Class Ġas ĠClass <* >) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġstart ( map : ĠMap < String , ĠString >) Ġ{ ĊĠĠĠĠĠĠĠ Ġthis . config Ġ= ĠNeo 4 j Sink Connector Config ( map ) ĊĠĠĠĠĠĠĠ Ġthis . neo 4 j Sink Service Ġ= ĠNeo 4 j Sink Service ( this . config ) ĊĠĠĠĠĠĠĠ Ġthis . error Service Ġ= ĠKafka Error Service ( this . config . kafka Broker Properties . as Properties (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠError Service . Error Config . from ( map . as Properties ()), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog :: error ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġput ( collection : ĠCollection < Sink Record >) Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( collection . isEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġdata Ġ= ĠEvent Builder () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. with Batch Size ( config . batch Size ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. with Sink Records ( collection ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. build () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġne o 4 j Sink Service . write Data ( data ) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch ( e : Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġerror Service . report ( collection . map Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠError Data ( it . topic (), Ġit . timestamp (), Ġit . key (), Ġit . value (), Ġit . kafka Partition (), Ġit . kafka Offset (), Ġthis :: class . java , Ġthis . config . database , Ġe ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġstop () Ġ{ ĊĠĠĠĠĠĠĠ Ġlog . info (" Stop () Ġ- ĠNeo 4 j ĠS ink ĠService ") ĊĠĠĠĠĠĠĠ ĠStreams Utils . ignore Exceptions ({ Ġne o 4 j Sink Service . close () Ġ}, ĠUn initialized Property AccessException :: class . java ) ĊĠĠĠ Ġ} Ċ }