package Ġch . source motion . vertx . k inesis . consumer . orche stra . consumer Ċ Ċ import Ġch . source motion . vertx . k inesis . consumer . orche stra .* Ċ import Ġch . source motion . vertx . k inesis . consumer . orche stra . consumer . Test Consumer Vert icle . Companion . RECORD _ SEND _ ADDR Ċ import Ġch . source motion . vertx . k inesis . consumer . orche stra . impl . EventBus Addr Ċ import Ġch . source motion . vertx . k inesis . consumer . orche stra . impl . Orche stra Cluster Name Ċ import Ġch . source motion . vertx . k inesis . consumer . orche stra . impl . Shard Id Ċ import Ġch . source motion . vertx . k inesis . consumer . orche stra . impl . codec . Local Codec Ċ import Ġch . source motion . vertx . k inesis . consumer . orche stra . impl . ext . ack Ċ import Ġch . source motion . vertx . k inesis . consumer . orche stra . impl . ext . completion Ċ import Ġch . source motion . vertx . k inesis . consumer . orche stra . impl . ext . isNotNull Or Blank Ċ import Ġch . source motion . vertx . k inesis . consumer . orche stra . impl . ext . shard Id Typed Ċ import Ġch . source motion . vertx . k inesis . consumer . orche stra . impl . resh arding . Merge Res h arding Event Ċ import Ġch . source motion . vertx . k inesis . consumer . orche stra . impl . resh arding . Res h arding Event Ċ import Ġch . source motion . vertx . k inesis . consumer . orche stra . impl . resh arding . Res h arding Type Ċ import Ġch . source motion . vertx . k inesis . consumer . orche stra . impl . resh arding . Split Res h arding Event Ċ import Ġch . source motion . vertx . k inesis . consumer . orche stra . testing .* Ċ import Ġch . source motion . vertx . k inesis . consumer . orche stra . testing . Shard Id Generator . resh arding Id Const ellation Ċ import Ġio . kot est . matchers . collections .* Ċ import Ġio . kot est . matchers . null s . should Not BeNull Ċ import Ġio . kot est . matchers . shouldBe Ċ import Ġio . kot est . matchers . types . shouldBe InstanceOf Ċ import Ġio . vertx . core . Vert x Ċ import Ġio . vertx . core . event bus . Delivery Options Ċ import Ġio . vertx . core . event bus . Reply Exception Ċ import Ġio . vertx . junit 5 . Vert x TestContext Ċ import Ġio . vertx . kotlin . coroutines . await Ċ import Ġkotlinx . coroutines . delay Ċ import Ġkotlinx . coroutines . launch Ċ import Ġorg . junit . jupiter . api . BeforeEach Ċ import Ġorg . junit . jupiter . api . Test Ċ import Ġsoftware . amazon . aw ssdk . core . Sdk Bytes Ċ import Ġsoftware . amazon . aw ssdk . services . k inesis . model . Record Ċ import Ġsoftware . amazon . aw ssdk . services . k inesis . model . Stream Description Ċ import Ġjava . time . Duration Ċ Ċ internal Ġclass ĠK inesis Consumer Vert icle Test Ġ: ĠAbstract K inesis And Redis Test () Ġ{ ĊĊĠĠĠ Ġ@ BeforeEach ĊĠĠĠ Ġinternal Ġfun ĠsetUp Record Message Codec () Ġ{ ĊĠĠĠĠĠĠĠ Ġevent Bus . register Default Codec ( Record :: class . java , ĠLocal Codec (" record - test - codec ")) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġinternal Ġfun Ġsingle _ shard _ consumer _ sun ny _ case ( vertx : ĠVert x , Ġtest Context : ĠVert x TestContext ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġrecord Batch ing Ġ= Ġ 1 0 Ġbatches Of Ġ 1 0 ĊĊĠĠĠĠĠĠĠ Ġtest Context . async Delayed ( record Batch ing . record Count ) Ġ{ Ġcheckpoint Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġstream Description Ġ= Ġk inesis Client . create AndGet Stream Description When Active ( 1 ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġvert x . event Bus (). consumer < Record >( RECORD _ SEND _ ADDR ) Ġ{ Ġmsg Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmsg . ack () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcheckpoint . flag () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdeploy Test Consumer Vert icle ( create K inesis Consumer Vert icle Config ( stream Description . getFirst Shard Id ())) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġk inesis Client . put Records ( record Batch ing ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġinternal Ġfun Ġconsumer _ retry _ on _ failure _ by _ configuration ( vertx : ĠVert x , Ġtest Context : ĠVert x TestContext ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġrecord Batch ing Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠĠ 1 0 Ġbatches Of Ġ 1 0 ĠaddTo Count Ġ 1 1 Ġ// Ġ+ Ġ 1 1 Ġbecause Ġeach Ġ 1 0 th Ġrecord Ġprocessing Ġwill Ġfail Ġand Ġretried Ġone Ġtime . ĊĊĠĠĠĠĠĠĠ Ġtest Context . async Delayed ( record Batch ing . record Count ) Ġ{ Ġcheckpoint Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġstream Description Ġ= Ġk inesis Client . create AndGet Stream Description When Active ( 1 ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġreceived Records Ġ= Ġ 0 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġlast Failed Record SequenceNumber : ĠString ? Ġ= Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvert x . event Bus (). consumer < Record >( RECORD _ SEND _ ADDR ) Ġ{ Ġmsg Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġrecord Ġ= Ġmsg . body () ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(++ received Records Ġ% Ġrecord Batch ing . records Per Batch Ġ== Ġ 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlast Failed Record SequenceNumber Ġ= Ġrecord . sequence Number () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmsg . reply Consume Record Failed Retry ( record ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠIf Ġthe Ġlast Ġfailed Ġsequence Ġnumber Ġis Ġpresent , Ġit 's Ġexpected Ġthat Ġthe Ġnext Ġrecord Ġis Ġthe Ġsame Ġhere , Ġas Ġretry Ġis Ġconfigured ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( last Failed Record SequenceNumber . isNotNull Or Blank ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtest Context . verify Ġ{ Ġlast Failed Record SequenceNumber . shouldBe ( record . sequence Number ()) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlast Failed Record SequenceNumber Ġ= Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmsg . ack () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info Ġ{ Ġ" Received Ġ$ received Records Ġrecords " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcheckpoint . flag () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}. completion (). await () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġdeploy Test Consumer Vert icle ( create K inesis Consumer Vert icle Config ( stream Description . getFirst Shard Id ())) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġk inesis Client . put Records ( record Batch ing ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġinternal Ġfun Ġconsumer _ retry _ on _ failure _ by _ exception _ override ( vertx : ĠVert x , Ġtest Context : ĠVert x TestContext ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġrecord Batch ing Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠĠ 1 0 Ġbatches Of Ġ 1 0 ĠaddTo Count Ġ 1 1 Ġ// Ġ+ Ġ 1 1 Ġbecause Ġeach Ġ 1 0 th Ġrecord Ġprocessing Ġwill Ġfail Ġand Ġretried Ġone Ġtime . ĊĊĠĠĠĠĠĠĠ Ġtest Context . async Delayed ( record Batch ing . record Count ) Ġ{ Ġcheckpoint Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġstream Description Ġ= Ġk inesis Client . create AndGet Stream Description When Active ( 1 ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġreceived Records Ġ= Ġ 0 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġlast Failed Record SequenceNumber : ĠString ? Ġ= Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvert x . event Bus (). consumer < Record >( RECORD _ SEND _ ADDR ) Ġ{ Ġmsg Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġrecord Ġ= Ġmsg . body () ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ(++ received Records Ġ% Ġrecord Batch ing . records Per Batch Ġ== Ġ 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlast Failed Record SequenceNumber Ġ= Ġrecord . sequence Number () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmsg . reply Consume Record Failed Retry ( record ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠIf Ġthe Ġlast Ġfailed Ġsequence Ġnumber Ġis Ġpresent , Ġit 's Ġexpected Ġthat Ġthe Ġnext Ġrecord Ġis Ġthe Ġsame Ġhere , Ġas Ġretry Ġis Ġconfigured ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( last Failed Record SequenceNumber . isNotNull Or Blank ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtest Context . verify Ġ{ Ġlast Failed Record SequenceNumber . shouldBe ( record . sequence Number ()) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlast Failed Record SequenceNumber Ġ= Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmsg . ack () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcheckpoint . flag () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}. completion (). await () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġdeploy Test Consumer Vert icle ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcreate K inesis Consumer Vert icle Config ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstream Description . getFirst Shard Id (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġerror Handling Ġ= ĠError Handling . IGNORE _ AND _ CONTINUE ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġk inesis Client . put Records ( record Batch ing ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġinternal Ġfun Ġconsumer _ ignore _ on _ failure _ by _ configuration ( vertx : ĠVert x , Ġtest Context : ĠVert x TestContext ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġrecord Batch ing Ġ= Ġ 1 0 Ġbatches Of Ġ 1 0 ĊĊĠĠĠĠĠĠĠ Ġtest Context . async Delayed ( record Batch ing . record Count ) Ġ{ Ġcheckpoint Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġstream Description Ġ= Ġk inesis Client . create AndGet Stream Description When Active ( 1 ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġrecord Sequence Numbers Ġ= Ġmutable ListOf < String >() ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġvert x . event Bus (). consumer < Record >( RECORD _ SEND _ ADDR ) Ġ{ Ġmsg Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġrecord Ġ= Ġmsg . body () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsequence Number Ġ= Ġrecord . sequence Number () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtest Context . verify Ġ{ Ġrecord Sequence Numbers . should Not Contain ( sequence Number ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrecord Sequence Numbers . add ( sequence Number ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠEach Ġrecord Ġprocessing Ġwill Ġfail ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmsg . fail ( 0 , Ġ"") ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcheckpoint . flag () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}. completion (). await () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġdeploy Test Consumer Vert icle ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcreate K inesis Consumer Vert icle Config ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstream Description . getFirst Shard Id (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġerror Handling Ġ= ĠError Handling . IGNORE _ AND _ CONTINUE , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfetch er Options Ġ= ĠFetch er Options ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġget Records Limit Ġ= Ġ 1 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrecords Fetch Interval Millis Ġ= Ġ 1 0 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdynamic Limit Adjustment Ġ= ĠDynamic Limit Adjustment ( enabled Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġk inesis Client . put Records ( record Batch ing ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠTest , Ġthat Ġthe Ġexception Ġhandling Ġbehavior Ġdefined Ġby Ġexception Ġoverrides Ġthe Ġoriginal Ġconfiguration . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġinternal Ġfun Ġconsumer _ ignore _ on _ failure _ exception _ override ( vertx : ĠVert x , Ġtest Context : ĠVert x TestContext ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġrecord Batch ing Ġ= Ġ 1 0 Ġbatches Of Ġ 1 0 ĊĊĠĠĠĠĠĠĠ Ġtest Context . async Delayed ( record Batch ing . record Count ) Ġ{ Ġcheckpoint Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġstream Description Ġ= Ġk inesis Client . create AndGet Stream Description When Active ( 1 ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġrecord Sequence Numbers Ġ= Ġmutable ListOf < String >() ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġvert x . event Bus (). consumer < Record >( RECORD _ SEND _ ADDR ) Ġ{ Ġmsg Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġrecord Ġ= Ġmsg . body () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsequence Number Ġ= Ġrecord . sequence Number () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtest Context . verify Ġ{ Ġrecord Sequence Numbers . should Not Contain ( sequence Number ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrecord Sequence Numbers . add ( sequence Number ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmsg . reply Consume Record Failed Ignore ( record ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcheckpoint . flag () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}. completion (). await () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġdeploy Test Consumer Vert icle ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcreate K inesis Consumer Vert icle Config ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstream Description . getFirst Shard Id (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġerror Handling Ġ= ĠError Handling . RETRY _ FROM _ FAILED _ RECORD , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfetch er Options Ġ= ĠFetch er Options ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġget Records Limit Ġ= Ġ 1 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrecords Fetch Interval Millis Ġ= Ġ 1 0 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdynamic Limit Adjustment Ġ= ĠDynamic Limit Adjustment ( enabled Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġk inesis Client . put Records ( record Batch ing ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠIf Ġthe Ġorche stra Ġis Ġconfigured Ġto Ġretry Ġfrom Ġfailed Ġrecord , Ġbut Ġa Ġfailure Ġ( Exception ) Ġnot Ġcontains Ġthe Ġnecessary ĊĠĠĠĠ Ġ* Ġinformation Ġ( record ), Ġthe Ġorche stra Ġwill Ġretry Ġbeginning Ġfrom Ġthe Ġprevious Ġiterator . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġinternal Ġfun Ġconsumer _ retry _ from _ failed _ configured _ but _ wrong _ exception ( ĊĠĠĠĠĠĠĠ Ġvert x : ĠVert x , ĊĠĠĠĠĠĠĠ Ġtest Context : ĠVert x TestContext ĊĠĠĠ Ġ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġrecord Batch ing Ġ= Ġ 1 Ġbatches Of Ġ 1 0 ĊĊĠĠĠĠĠĠĠ Ġtest Context . async ( record Batch ing . record Count ) Ġ{ Ġcheckpoint Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġstream Description Ġ= Ġk inesis Client . create AndGet Stream Description When Active ( 1 ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġrecord SequenceNumber : ĠString ? Ġ= Ġnull ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġvert x . event Bus (). consumer < Record >( RECORD _ SEND _ ADDR ) Ġ{ Ġmsg Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠEvery Ġtime Ġthe Ġsame Ġ/ Ġfirst Ġrecord Ġis Ġexcept ed ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġrecord Ġ= Ġmsg . body () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( record SequenceNumber . isNotNull Or Blank ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtest Context . verify Ġ{ Ġrecord SequenceNumber . shouldBe ( record . sequence Number ()) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrecord SequenceNumber Ġ= Ġrecord . sequence Number () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠEach Ġrecord Ġprocessing Ġwill Ġfail , Ġbut ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmsg . fail ( 0 , Ġ" Test Ġfailure ") ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcheckpoint . flag () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}. completion (). await () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġdeploy Test Consumer Vert icle ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcreate K inesis Consumer Vert icle Config ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstream Description . getFirst Shard Id (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġerror Handling Ġ= ĠError Handling . RETRY _ FROM _ FAILED _ RECORD , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfetch er Options Ġ= ĠFetch er Options ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrecords Fetch Interval Millis Ġ= Ġ 1 0 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdynamic Limit Adjustment Ġ= ĠDynamic Limit Adjustment ( enabled Ġ= Ġfalse ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġk inesis Client . put Records ( record Batch ing ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġinternal Ġfun ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsplit _ resh arding ( test Context : ĠVert x TestContext ) Ġ= Ġtest Context . async Delayed ( 1 ) Ġ{ Ġcheckpoint Ġ-> ĊĠĠĠĠĠĠĠ Ġval Ġ( parent Shard Id , Ġfirst Child , Ġsecond Child ) Ġ= Ġre sharding Id Const ellation () ĊĠĠĠĠĠĠĠ Ġval Ġstream Description Ġ= Ġk inesis Client . create AndGet Stream Description When Active ( 1 ) ĊĊĠĠĠĠĠĠĠ Ġdeploy Test Consumer Vert icle ( create K inesis Consumer Vert icle Config ( parent Shard Id )) ĊĊĠĠĠĠĠĠĠ Ġevent Bus . consumer < Res h arding Event >( EventBus Addr . resh arding . notification ) Ġ{ Ġmsg Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġevent Ġ= Ġmsg . body () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtest Context . verify Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġevent . resh arding Type . shouldBe ( Res h arding Type . SPLIT ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġevent . shouldBe InstanceOf < Split Res h arding Event >() ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġevent . finished Parent Shard Id . shouldBe ( parent Shard Id ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġevent . child Shard Ids . should Contain Exactly In Any Order ( firstChild , Ġsecond Child ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcheckpoint . flag () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġk inesis Client . split Shard F air ( stream Description . shards (). first ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġinternal Ġfun Ġmerge _ resh arding ( test Context : ĠVert x TestContext ) Ġ= Ġtest Context . async ( 2 ) Ġ{ Ġcheckpoint Ġ-> ĊĠĠĠĠĠĠĠ Ġval Ġ( parent Shard Id , Ġadjacent Parent Shard Id , Ġchild Shard Id ) Ġ= Ġre sharding Id Const ellation () ĊĠĠĠĠĠĠĠ Ġval Ġparent Shard Ids Ġ= ĠlistOf ( parent Shard Id , Ġadjacent Parent Shard Id ). to Mutable List () ĊĊĠĠĠĠĠĠĠ Ġval Ġstream Description Ġ= Ġk inesis Client . create AndGet Stream Description When Active ( 2 ) ĊĊĠĠĠĠĠĠĠ Ġstream Description . shards (). forEach Ġ{ Ġshard Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġdeploy Test Consumer Vert icle ( create K inesis Consumer Vert icle Config ( shard . shard Id Typed ())) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġevent Bus . consumer < Res h arding Event >( EventBus Addr . resh arding . notification ) Ġ{ Ġmsg Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġevent Ġ= Ġmsg . body () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtest Context . verify Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġevent . resh arding Type . shouldBe ( Res h arding Type . MERGE ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġevent . shouldBe InstanceOf < Merge Res h arding Event >() ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġfinished Parent Shard Id Ġ= Ġevent . finished Parent Shard Id ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġparent Shard Ids . should Contain ( finished Parent Shard Id ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġparent Shard Ids . remove ( finished Parent Shard Id ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġevent . child Shard Id . shouldBe ( child Shard Id ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcheckpoint . flag () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġk inesis Client . merge Shards ( stream Description . shards ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠSim ulates Ġthe Ġrestart Ġof Ġa Ġconsumer Ġto Ġtest Ġinteraction Ġwith Ġiterator Ġpersistence Ġetc . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġinternal Ġfun Ġconsumer _ restart ( test Context : ĠVert x TestContext ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġrecord Batch ing Ġ= Ġ 1 Ġbatches Of Ġ 1 0 ĊĠĠĠĠĠĠĠ Ġval Ġrecord Count Ġ= Ġrecord Batch ing . record Count ĊĊĠĠĠĠĠĠĠ Ġtest Context . async Delayed ( record Count Ġ* Ġ 2 /* ĠWe Ġput Ġthe Ġdouble Ġamount Ġof Ġrecords */) Ġ{ Ġcheckpoint Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġstream Description Ġ= Ġk inesis Client . create AndGet Stream Description When Active ( 1 ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWe Ġneed Ġa Ġcontinued Ġindexing Ġof Ġrecords , Ġso Ġwe Ġare Ġable Ġto Ġidentify Ġeach Ġround Ġthe Ġconsumer Ġwill Ġrun . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġput Record Idx Ġ= Ġ 0 ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġgenerate Record Data Ġ= Ġ{ ĠS dk Bytes . from Utf 8 String ("${ put Record Idx ++ }") Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġconsumer Deployment Id : ĠString ? Ġ= Ġnull ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġconsumer Round Starter Ġ= Ġsuspend Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer Deployment Id Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdeploy Test Consumer Vert icle ( create K inesis Consumer Vert icle Config ( stream Description . getFirst Shard Id ())) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġreceived Record Indices Ġ= Ġmutable ListOf < Int >() ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġvert x . event Bus (). consumer < Record >( RECORD _ SEND _ ADDR ) Ġ{ Ġmsg Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġrecord Idx Ġ= Ġmsg . body (). data (). as Utf 8 String (). toInt () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( received Record Indices . isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrecord Idx . shouldBe ( received Record Indices . last () Ġ+ Ġ 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreceived Record Indices . add ( record Idx ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmsg . ack () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( received Record Indices . size Ġ== Ġrecord Count ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info Ġ{ Ġ" Received Ġfirst Ġbunch Ġof Ġrecords . ĠRestart Ġconsumer " Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdefault Test Scope . launch Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtest Context . verify Ġ{ Ġconsumer Deployment Id . should Not BeNull () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvert x . un deploy ( consumer Deployment Id !! ). await () ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer Round Starter () ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġk inesis Client . put Records ( record Batch ing , Ġrecord Data Supplier Ġ= Ġ{ Ġgenerate Record Data () Ġ}) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠVerification Ġof Ġthe Ġsecond Ġrun Ġ/ Ġdeployment Ġof Ġthe Ġconsumer Ġvert icle . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠSo Ġwe Ġtest Ġiterator Ġpersistence Ġworks Ġas Ġexpected ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( received Record Indices . size Ġ== Ġrecord Count Ġ* Ġ 2 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtest Context . verify Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrepeat ( record Count ) Ġ{ Ġidx Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġexpected Idx Ġ= Ġidx Ġ+ Ġrecord Count ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreceived Record Indices . should Contain ( expected Idx ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcheckpoint . flag () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}. completion (). await () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer Round Starter () ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġk inesis Client . put Records ( record Batch ing , Ġrecord Data Supplier Ġ= Ġ{ Ġgenerate Record Data () Ġ}) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġinternal Ġfun Ġshard _ progress _ flag ging ( test Context : ĠVert x TestContext ) Ġ= Ġtest Context . async Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġshard Progress Expiration Millis Ġ= Ġ 1 0 0 L ĊĠĠĠĠĠĠĠ Ġval Ġshard Id Ġ= Ġk inesis Client . create AndGet Stream Description When Active ( 1 ). getFirst Shard Id () ĊĠĠĠĠĠĠĠ Ġval Ġdeployment Id Ġ= Ġdeploy Test Consumer Vert icle ( create K inesis Consumer Vert icle Config ( shard Id , Ġshard Progress Expiration Millis Ġ= Ġshard Progress Expiration Millis )) ĊĠĠĠĠĠĠĠ Ġshard State Persistence Service . get Shard Ids InProgress (). should Contain Exactly ( shard Id ) ĊĠĠĠĠĠĠĠ Ġdelay ( shard Progress Expiration Millis Ġ* Ġ 2 ) ĊĠĠĠĠĠĠĠ Ġshard State Persistence Service . get Shard Ids InProgress (). should Contain Exactly ( shard Id ) ĊĠĠĠĠĠĠĠ Ġvert x . un deploy ( deployment Id ). await () ĊĠĠĠĠĠĠĠ Ġshard State Persistence Service . get Shard Ids InProgress (). shouldBe Empty () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun ĠStream Description . getFirst Shard Id () Ġ= Ġsh ards (). first (). shard Id Typed () ĊĊĠĠĠ Ġprivate Ġsuspend Ġfun Ġdeploy Test Consumer Vert icle ( ĊĠĠĠĠĠĠĠ Ġoptions : ĠK inesis Consumer Vert icle Options , ĊĠĠĠĠĠĠĠ Ġinstances : ĠInt Ġ= Ġ 1 ĊĠĠĠ Ġ) Ġ= Ġdeploy Test Vert icle < Test Consumer Vert icle >( options , Ġinstances ) ĊĊĠĠĠ Ġprivate Ġfun Ġcreate K inesis Consumer Vert icle Config ( ĊĠĠĠĠĠĠĠ Ġshard Id : ĠSh ard Id , ĊĠĠĠĠĠĠĠ Ġshard Iterator Strategy : ĠSh ard Iterator Strategy Ġ= ĠSh ard Iterator Strategy . EXIST ING _ OR _ LATE ST , ĊĠĠĠĠĠĠĠ Ġerror Handling : ĠError Handling Ġ= ĠError Handling . RETRY _ FROM _ FAILED _ RECORD , ĊĠĠĠĠĠĠĠ Ġshard Progress Expiration Millis : ĠLong Ġ= ĠVert x K inesis Orche stra Options . DEFAULT _ SH ARD _ PROGRESS _ EXPI RATION _ M ILL IS , ĊĠĠĠĠĠĠĠ Ġfetch er Options : ĠFetch er Options Ġ= ĠFetch er Options () ĊĠĠĠ Ġ) Ġ= ĠK inesis Consumer Vert icle Options ( ĊĠĠĠĠĠĠĠ Ġshard Id , ĊĠĠĠĠĠĠĠ ĠO rche stra Cluster Name ( TEST _ APPLICATION _ NAME , ĠTEST _ STREAM _ NAME ), ĊĠĠĠĠĠĠĠ Ġshard Iterator Strategy , ĊĠĠĠĠĠĠĠ Ġerror Handling , ĊĠĠĠĠĠĠĠ Ġshard Progress Expiration Millis Ġ= Ġshard Progress Expiration Millis , ĊĠĠĠĠĠĠĠ Ġfetch er Options Ġ= Ġfetch er Options ĊĠĠĠ Ġ) Ċ } Ċ Ċ class ĠTest Consumer Vert icle Ġ: ĠAbstract K inesis Consumer Coroutine Vert icle () Ġ{ ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġconst Ġval ĠREC ORD _ SEND _ ADDR Ġ= Ġ"/ k inesis - consumer - orche ster / testing / single - record " ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġsuspend Ġfun Ġon Records Async ( records : ĠList < Record >) Ġ{ ĊĠĠĠĠĠĠĠ Ġrun Catch ing Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrecords . forEach Ġ{ Ġrecord Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvert x . event Bus (). request < Unit >( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠREC ORD _ SEND _ ADDR , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrecord , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠEnsure Ġsend Ġwill Ġnot Ġtimeout , Ġeven Ġon Ġlonger Ġrunning Ġtests ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠDelivery Options (). set Send Timeout ( Duration . of Minutes ( 1 0 ). to Millis ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ). await () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ}. exception OrNull ()?. let Ġ{ Ġthrowable Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġif Ġ( throwable Ġis ĠReply Exception Ġ&& Ġthrowable . is K inesis Consumer Exception ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrowable . to K inesis Consumer Exception () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrowable ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ