package Ġch . source motion . vertx . k inesis . consumer . orche stra . consumer Ċ Ċ import Ġch . source motion . vertx . k inesis . consumer . orche stra . Shard Iterator Strategy Ċ import Ġch . source motion . vertx . k inesis . consumer . orche stra . impl . SequenceNumber Ċ import Ġch . source motion . vertx . k inesis . consumer . orche stra . impl . SequenceNumber Iterator Position Ċ import Ġch . source motion . vertx . k inesis . consumer . orche stra . impl . Shard Iterator Ċ import Ġch . source motion . vertx . k inesis . consumer . orche stra . spi . Shard State Persistence Service Async Ċ import Ġch . source motion . vertx . k inesis . consumer . orche stra . testing . Abstract Vert x Test Ċ import Ġch . source motion . vertx . k inesis . consumer . orche stra . testing . Shard Id Generator Ċ import Ġch . source motion . vertx . k inesis . consumer . orche stra . testing . TEST _ CLUSTER _ OR CHE STR A _ NAME Ċ import Ġcom . n ha ar man . mock it ok otlin 2 . any Ċ import Ġcom . n ha ar man . mock it ok otlin 2 . do Return Ċ import Ġcom . n ha ar man . mock it ok otlin 2 . eq Ċ import Ġcom . n ha ar man . mock it ok otlin 2 . mock Ċ import Ġio . kot est . matchers . null s . shouldBe Null Ċ import Ġio . kot est . matchers . shouldBe Ċ import Ġio . vertx . junit 5 . Vert x TestContext Ċ import Ġorg . junit . jupiter . api . Test Ċ import Ġsoftware . amazon . aw ssdk . services . k inesis . K inesis Async Client Ċ import Ġsoftware . amazon . aw ssdk . services . k inesis . model . Get Shard Iterator Request Ċ import Ġsoftware . amazon . aw ssdk . services . k inesis . model . Get Shard Iterator Response Ċ import Ġjava . util . concurrent . Comp letableFuture Ċ import Ġjava . util . function . Consumer Ċ Ċ internal Ġclass ĠStart Fetch Position Lookup Test Ġ: ĠAbstract Vert x Test () Ġ{ ĊĠĠĠ Ġprivate Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġconst Ġval ĠTEST _ STREAM _ NAME Ġ= Ġ" test - stream " ĊĠĠĠĠĠĠĠ Ġconst Ġval ĠCON SUM ER _ INFO Ġ= Ġ" test - consumer - info " ĊĠĠĠĠĠĠĠ Ġval Ġshard Id Ġ= ĠSh ard Id Generator . generate Shard Id () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġinternal Ġfun Ġget _ start _ fetch position _ force _ latest ( test Context : ĠVert x TestContext ) Ġ= Ġasync Test ( test Context ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġexpected Shard Iterator Ġ= Ġ" next - shard - iterator " ĊĠĠĠĠĠĠĠ Ġval Ġk inesis Client Ġ= Ġmock K inesis Client ( expected Shard Iterator ) ĊĠĠĠĠĠĠĠ Ġval Ġsut Ġ= ĠStart Fetch Position Lookup ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvert x , ĊĠĠĠĠĠĠĠĠĠĠĠ ĠCON SUM ER _ INFO , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġshard Id , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcreate Force Latest Shard Iterator Options (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmock (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġk inesis Client ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġval Ġfetch Position Ġ= Ġsut . getStart Fetch Position () ĊĊĠĠĠĠĠĠĠ Ġfetch Position . iterator . shouldBe ( Shard Iterator ( expected Shard Iterator )) ĊĠĠĠĠĠĠĠ Ġfetch Position . sequence Number . shouldBe Null () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġinternal Ġfun Ġget _ start _ fetch position _ existing _ or _ latest _ existing ( test Context : ĠVert x TestContext ) Ġ= Ġasync Test ( test Context ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġexpected Shard Iterator Ġ= Ġ" next - shard - iterator " ĊĠĠĠĠĠĠĠ Ġval Ġexpected Sequence Nbr Value Ġ= Ġ" b 9 9 dc 8 6 6 - 0 fc 3 - 4 d 3 6 - 9 cf 1 - c 8 1 d 9 1 7 6 c 8 2 1 " ĊĠĠĠĠĠĠĠ Ġval Ġexpected Sequence Nbr Iterator Position Ġ= ĠSequence Number Iterator Position . AFTER ĊĠĠĠĠĠĠĠ Ġval Ġexpected SequenceNumber Ġ= ĠSequence Number ( expected Sequence Nbr Value , Ġexpected Sequence Nbr Iterator Position ) ĊĊĠĠĠĠĠĠĠ Ġval Ġk inesis Client Ġ= Ġmock K inesis Client ( expected Shard Iterator ) ĊĊĠĠĠĠĠĠĠ Ġval Ġshard State Persistence Service Ġ= Ġmock < Shard State Persistence Service Async > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Blocking Ġ{ Ġget Consumer Shard SequenceNumber ( eq ( shard Id )) Ġ} Ġdo Return Ġexpected SequenceNumber ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġsut Ġ= ĠStart Fetch Position Lookup ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvert x , ĊĠĠĠĠĠĠĠĠĠĠĠ ĠCON SUM ER _ INFO , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġshard Id , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcreate Existing Or Latest Shard Iterator Options (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġshard State Persistence Service , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġk inesis Client , ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġval Ġfetch Position Ġ= Ġsut . getStart Fetch Position () ĊĊĠĠĠĠĠĠĠ Ġfetch Position . iterator . shouldBe ( Shard Iterator ( expected Shard Iterator )) ĊĠĠĠĠĠĠĠ Ġfetch Position . sequence Number . shouldBe ( expected SequenceNumber ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġinternal Ġfun Ġget _ start _ fetch position _ existing _ or _ latest _ latest ( test Context : ĠVert x TestContext ) Ġ= Ġasync Test ( test Context ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġexpected Shard Iterator Ġ= Ġ" next - shard - iterator " ĊĊĠĠĠĠĠĠĠ Ġval Ġk inesis Client Ġ= Ġmock K inesis Client ( expected Shard Iterator ) ĊĊĠĠĠĠĠĠĠ Ġval Ġshard State Persistence Service Ġ= Ġmock < Shard State Persistence Service Async > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Blocking Ġ{ Ġget Consumer Shard SequenceNumber ( eq ( shard Id )) Ġ} Ġdo Return Ġnull ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġsut Ġ= ĠStart Fetch Position Lookup ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvert x , ĊĠĠĠĠĠĠĠĠĠĠĠ ĠCON SUM ER _ INFO , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġshard Id , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcreate Existing Or Latest Shard Iterator Options (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġshard State Persistence Service , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġk inesis Client , ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġval Ġfetch Position Ġ= Ġsut . getStart Fetch Position () ĊĊĠĠĠĠĠĠĠ Ġfetch Position . iterator . shouldBe ( Shard Iterator ( expected Shard Iterator )) ĊĠĠĠĠĠĠĠ Ġfetch Position . sequence Number . shouldBe Null () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġinternal Ġfun Ġget _ start _ fetch position _ existing _ or _ latest _ import ( test Context : ĠVert x TestContext ) Ġ= Ġasync Test ( test Context ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġexpected Shard Iterator Ġ= Ġ" next - shard - iterator " ĊĊĠĠĠĠĠĠĠ Ġval Ġk inesis Client Ġ= Ġmock K inesis Client ( expected Shard Iterator ) ĊĊĠĠĠĠĠĠĠ Ġval Ġshard State Persistence Service Ġ= Ġmock < Shard State Persistence Service Async > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Blocking Ġ{ Ġget Consumer Shard SequenceNumber ( eq ( shard Id )) Ġ} Ġdo Return Ġnull ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġsut Ġ= ĠStart Fetch Position Lookup ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvert x , ĊĠĠĠĠĠĠĠĠĠĠĠ ĠCON SUM ER _ INFO , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġshard Id , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcreate Existing Or Latest Shard Iterator Options (), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġshard State Persistence Service , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġk inesis Client , ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġval Ġfetch Position Ġ= Ġsut . getStart Fetch Position () ĊĊĠĠĠĠĠĠĠ Ġfetch Position . iterator . shouldBe ( Shard Iterator ( expected Shard Iterator )) ĊĠĠĠĠĠĠĠ Ġfetch Position . sequence Number . shouldBe Null () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġmock K inesis Client ( expected Shard Iterator : ĠString ): ĠK inesis Async Client Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġget Response Ġ= Ġmock < Get Shard Iterator Response > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġshard Iterator () Ġ} Ġdo Return Ġexpected Shard Iterator ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġmock Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġget Shard Iterator ( any < Consumer < Get Shard Iterator Request . Builder >> ()) Ġ} Ġdo Return ĠCompletableFuture . completed Future ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġget Response ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcreate Force Latest Shard Iterator Options () Ġ= Ġmock < K inesis Consumer Vert icle Options > Ġ{ ĊĠĠĠĠĠĠĠ Ġon Ġ{ Ġcluster Name Ġ} Ġdo Return ĠTEST _ CLUSTER _ OR CHE STR A _ NAME ĊĠĠĠĠĠĠĠ Ġon Ġ{ Ġshard Iterator Strategy Ġ} Ġdo Return ĠSh ard Iterator Strategy . FORCE _ LATE ST ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcreate Existing Or Latest Shard Iterator Options ( sequence Nbr Import Addr : ĠString ? Ġ= Ġnull ) Ġ= ĊĠĠĠĠĠĠĠ Ġmock < K inesis Consumer Vert icle Options > Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġcluster Name Ġ} Ġdo Return ĠTEST _ CLUSTER _ OR CHE STR A _ NAME ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġshard Iterator Strategy Ġ} Ġdo Return ĠSh ard Iterator Strategy . EXIST ING _ OR _ LATE ST ĊĠĠĠĠĠĠĠĠĠĠĠ Ġon Ġ{ Ġthis . sequence Nbr Import Address Ġ} Ġdo Return Ġsequence Nbr Import Addr ĊĠĠĠĠĠĠĠ Ġ} Ċ } Ċ