package Ġintegr ations . kafka Ċ Ċ import Ġintegr ations . kafka . Kafka TestUtils . create Consumer Ċ import Ġkotlinx . coroutines . run Blocking Ċ import Ġorg . apache . kafka . clients . consumer . Consumer Config Ċ import Ġorg . apache . kafka . clients . producer . Producer Record Ċ import Ġorg . apache . kafka . common . Topic Partition Ċ import Ġorg . hamcrest . Matchers Ċ import Ġorg . junit . Test Ċ import Ġorg . neo 4 j . function . Throw ing Supplier Ċ import Ġorg . neo 4 j . kernel . internal . Graph Database API Ċ import Ġorg . neo 4 j . test . assertion . Assert Ċ import Ġstreams . serialization . JSON Utils Ċ import Ġjava . util .* Ċ import Ġjava . util . concurrent . TimeUnit Ċ Ċ class ĠKafka Event Sink Commit Ġ: ĠKafka Event Sink Base () Ġ{ ĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġshould Write Last Offset With No Auto Commit () Ġ= Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġtopic Ġ= ĠUUID . randomUUID (). toString () ĊĠĠĠĠĠĠĠ Ġgraph Database Builder . set Config (" streams . sink . topic . cy pher .$ topic ", Ġcy pher Query Template ) ĊĠĠĠĠĠĠĠ Ġgraph Database Builder . set Config (" kafka .${ Consumer Config . ENABLE _ AUTO _ COMMIT _ CONFIG }", Ġ" false ") ĊĠĠĠĠĠĠĠ Ġdb Ġ= Ġgraph Database Builder . new Graph Database () Ġas ĠGraph Database API ĊĠĠĠĠĠĠĠ Ġval Ġpartition Ġ= Ġ 0 ĊĠĠĠĠĠĠĠ Ġvar Ġproducer Record Ġ= ĠProducer Record ( topic , Ġpartition , ĠUUID . randomUUID (). toString (), ĠJSON Utils . writeValue As Bytes ( data )) ĊĠĠĠĠĠĠĠ Ġkafka Producer . send ( producer Record ). get () ĊĠĠĠĠĠĠĠ Ġval ĠnewData Ġ= Ġdata . to Mutable Map () ĊĠĠĠĠĠĠĠ ĠnewData [" id "] Ġ= Ġ 2 ĊĠĠĠĠĠĠĠ Ġproducer Record Ġ= ĠProducer Record ( topic , Ġpartition , ĠUUID . randomUUID (). toString (), ĠJSON Utils . writeValue As Bytes ( new Data )) ĊĠĠĠĠĠĠĠ Ġval Ġresp Ġ= Ġkafka Producer . send ( producer Record ). get () ĊĊĠĠĠĠĠĠĠ ĠAssert . assert Event ually ( Throw ing Supplier < Boolean , ĠException > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġquery Ġ= Ġ" MATCH Ġ( n : Label ) ĠRETURN Ġcount (*) ĠAS Ġcount " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġdb . execute ( query ). column As < Long >(" count ") ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġkafka Consumer Ġ= Ġcreate Consumer < String , ĠByteArray >( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġkafka Ġ= ĠKafka Event Sink Suite IT . kafka , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġschema Registry Ġ= ĠKafka Event Sink Suite IT . schema Registry ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġoffset And Metadata Ġ= Ġkafka Consumer . committed ( Topic Partition ( topic , Ġpartition )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġkafka Consumer . close () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult . hasNext () Ġ&& Ġresult . next () Ġ== Ġ 2 L Ġ&& Ġ! result . hasNext () Ġ&& Ġresp . offset () Ġ+ Ġ 1 Ġ== Ġoffset And Metadata . offset () ĊĠĠĠĠĠĠĠ Ġ}, ĠMatch ers . equalTo ( true ), Ġ 3 0 , ĠTimeUnit . SECONDS ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġshould Write Last Offset With Async Commit () Ġ= Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġtopic Ġ= ĠUUID . randomUUID (). toString () ĊĠĠĠĠĠĠĠ Ġgraph Database Builder . set Config (" streams . sink . topic . cy pher .$ topic ", Ġcy pher Query Template ) ĊĠĠĠĠĠĠĠ Ġgraph Database Builder . set Config (" kafka .${ Consumer Config . ENABLE _ AUTO _ COMMIT _ CONFIG }", Ġ" false ") ĊĠĠĠĠĠĠĠ Ġgraph Database Builder . set Config (" kafka . streams . commit . async ", Ġ" true ") ĊĠĠĠĠĠĠĠ Ġdb Ġ= Ġgraph Database Builder . new Graph Database () Ġas ĠGraph Database API ĊĠĠĠĠĠĠĠ Ġval Ġpartition Ġ= Ġ 0 ĊĠĠĠĠĠĠĠ Ġvar Ġproducer Record Ġ= ĠProducer Record ( topic , Ġpartition , ĠUUID . randomUUID (). toString (), ĠJSON Utils . writeValue As Bytes ( data )) ĊĠĠĠĠĠĠĠ Ġkafka Producer . send ( producer Record ). get () ĊĠĠĠĠĠĠĠ Ġval ĠnewData Ġ= Ġdata . to Mutable Map () ĊĠĠĠĠĠĠĠ ĠnewData [" id "] Ġ= Ġ 2 ĊĠĠĠĠĠĠĠ Ġproducer Record Ġ= ĠProducer Record ( topic , Ġpartition , ĠUUID . randomUUID (). toString (), ĠJSON Utils . writeValue As Bytes ( new Data )) ĊĠĠĠĠĠĠĠ Ġval Ġresp Ġ= Ġkafka Producer . send ( producer Record ). get () ĊĊĠĠĠĠĠĠĠ ĠAssert . assert Event ually ( Throw ing Supplier < Boolean , ĠException > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġquery Ġ= Ġ" MATCH Ġ( n : Label ) ĠRETURN Ġcount (*) ĠAS Ġcount " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġdb . execute ( query ). column As < Long >(" count ") ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġkafka Consumer Ġ= Ġcreate Consumer < String , ĠByteArray >( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġkafka Ġ= ĠKafka Event Sink Suite IT . kafka , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġschema Registry Ġ= ĠKafka Event Sink Suite IT . schema Registry ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġoffset And Metadata Ġ= Ġkafka Consumer . committed ( Topic Partition ( topic , Ġpartition )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġkafka Consumer . close () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult . hasNext () Ġ&& Ġresult . next () Ġ== Ġ 2 L Ġ&& Ġ! result . hasNext () Ġ&& Ġresp . offset () Ġ+ Ġ 1 Ġ== Ġoffset And Metadata ?. offset () ĊĠĠĠĠĠĠĠ Ġ}, ĠMatch ers . equalTo ( true ), Ġ 3 0 , ĠTimeUnit . SECONDS ) ĊĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` should Ġfix Ġissue Ġ 1 8 6 Ġwith Ġauto Ġcommit Ġfalse `() Ġ= Ġrun Blocking Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġproduct Ġ= Ġ" product " Ġto Ġ" MERGE Ġ( p : Product Ġ{ id : Ġevent . id }) ĠON ĠCREATE ĠSET Ġp . name Ġ= Ġevent . name " ĊĠĠĠĠĠĠĠ Ġval Ġcustomer Ġ= Ġ" customer " Ġto Ġ" MERGE Ġ( c : Customer Ġ{ id : Ġevent . id }) ĠON ĠCREATE ĠSET Ġc . name Ġ= Ġevent . name " ĊĠĠĠĠĠĠĠ Ġval Ġb ought Ġ= Ġ" b ought " Ġto Ġ""" ĊĠĠĠĠĠĠĠĠĠĠĠ ĠMER GE Ġ( c : Customer Ġ{ id : Ġevent . id }) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠMER GE Ġ( p : Product Ġ{ id : Ġevent . id }) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠMER GE Ġ( c )- [: BO UG HT ]-> ( p ) ĊĠĠĠĠĠĠĠ Ġ""" . trim Indent () ĊĠĠĠĠĠĠĠ Ġgraph Database Builder . set Config (" streams . sink . topic . cy pher .${ product . first }", Ġproduct . second ) ĊĠĠĠĠĠĠĠ Ġgraph Database Builder . set Config (" streams . sink . topic . cy pher .${ customer . first }", Ġcustomer . second ) ĊĠĠĠĠĠĠĠ Ġgraph Database Builder . set Config (" streams . sink . topic . cy pher .${ b ought . first }", Ġb ought . second ) ĊĠĠĠĠĠĠĠ Ġgraph Database Builder . set Config (" kafka .${ Consumer Config . ENABLE _ AUTO _ COMMIT _ CONFIG }", Ġ" false ") ĊĠĠĠĠĠĠĠ Ġdb Ġ= Ġgraph Database Builder . new Graph Database () Ġas ĠGraph Database API ĊĊĠĠĠĠĠĠĠ Ġval Ġprops Ġ= Ġmap Of (" id " Ġto Ġ 1 , Ġ" name " Ġto Ġ" My ĠAwesome ĠProduct ") ĊĠĠĠĠĠĠĠ Ġvar Ġproducer Record Ġ= ĠProducer Record ( product . first , ĠUUID . randomUUID (). toString (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠJSON Utils . writeValue As Bytes ( props )) ĊĠĠĠĠĠĠĠ Ġkafka Producer . send ( producer Record ). get () ĊĠĠĠĠĠĠĠ ĠAssert . assert Event ually ( Throw ing Supplier < Boolean , ĠException > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġquery Ġ= Ġ""" ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠMATCH Ġ( p : Product ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠWHERE Ġproperties ( p ) Ġ= Ġ{ props } ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠRETURN Ġcount ( p ) ĠAS Ġcount ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ""" . trim Indent () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġdb . execute ( query , Ġmap Of (" props " Ġto Ġprops )). column As < Long >(" count ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult . hasNext () Ġ&& Ġresult . next () Ġ== Ġ 1 L Ġ&& Ġ! result . hasNext () ĊĠĠĠĠĠĠĠ Ġ}, ĠMatch ers . equalTo ( true ), Ġ 3 0 , ĠTimeUnit . SECONDS ) ĊĠĠĠ Ġ} Ċ }