package Ġch . source motion . vertx . k inesis . consumer . orche stra . spi Ċ Ċ import Ġch . source motion . vertx . k inesis . consumer . orche stra . impl .* Ċ import Ġio . vertx . codegen . annotations . Proxy Gen Ċ import Ġio . vertx . core . AsyncResult Ċ import Ġio . vertx . core . Handler Ċ import Ġio . vertx . core . json . JsonObject Ċ import Ġio . vertx . kotlin . coroutines . await Result Ċ Ċ /** Ċ Ġ* ĠThis Ġis Ġthe Ġentry Ġpoint Ġto Ġcustomize Ġhow Ġthe Ġshard Ġiterators Ġ/ Ġconsumers Ġstates Ġare Ġpersisted Ċ Ġ* Ċ Ġ* ĠThe Ġ[ Shard State Persistence Service ] Ġis Ġresponsible Ġto : Ċ Ġ* Ġ- ĠPers ist Ġthe Ġshard Ġiterator Ġof Ġeach Ġconsumer . Ċ Ġ* Ġ- ĠThe Ġstate Ġof Ġeach Ġconsumer Ġis Ġon Ġd uty , Ġor Ġwhich Ġsh ards Ġare Ġcurrently Ġconsumed . Ċ Ġ* Ġ- ĠPersistence Ġof Ġthe ĠK inesis Ġshard Ġiterator Ġof Ġeach Ġconsumed Ġshard . Ċ Ġ* Ġ- ĠThe Ġcount Ġof Ġdone Ġmerge Ġparent Ġshard Ġconsumers Ġof Ġthe Ġresulting Ġchild Ġshard . Ċ Ġ* Ġ- ĠKeep Ġalive Ġof Ġshard Ġprogress Ġto Ġavoid Ġa Ġshard Ġget Ġconsumed Ġby Ġmore Ġthan Ġone Ġconsumer Ġat Ġonce . Ċ Ġ* Ċ Ġ* ĠTo Ġexpose Ġyour Ġown Ġimplementation Ġplease Ġuse Ġ[ Shard State Persistence Service Factory . expose ] Ċ Ġ*/ Ċ @ Proxy Gen Ċ interface ĠSh ard State Persistence Service Ġ{ ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* Ġ@ return ĠList Ġof Ġshard Ġids Ġthey Ġare Ġcurrently Ġin Ġprogress . ĠMe ans Ġthey Ġare Ġfetched Ġby Ġa Ġconsumer Ġof Ġan Ġorche stra Ġinstance . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġget Shard Ids InProgress ( handler : ĠHandler < AsyncResult < List < String >> >) ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠFlags Ġa Ġshard Ġas Ġcurrently Ġfetched Ġby Ġconsumer Ġof Ġan Ġorche stra Ġinstance Ġand Ġso Ġin Ġprogress . ĠThe Ġflag Ġshould ĊĠĠĠĠ Ġ* Ġexpire Ġaccording Ġto Ġ[ ch . source motion . vertx . k inesis . consumer . orche stra . Vert x K inesis Orche stra Options . shard Progress Expiration ]. ĊĠĠĠĠ Ġ* ĠThe Ġexpiration Ġshould Ġprevent Ġthe Ġorche stration Ġfrom Ġdeath Ġlocks Ġif Ġan Ġorche stra Ġinstance Ġwas Ġnot Ġgrace fully ĊĠĠĠĠ Ġ* Ġstopped . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġflag Shard InProgress ( shard Id : ĠString , Ġhandler : ĠHandler < AsyncResult < Boolean >>) ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠRemoves Ġa Ġthe Ġflag Ġthat 's Ġa Ġshard Ġis Ġcurrently Ġfetched Ġby Ġa Ġconsumer Ġof Ġan Ġorche stra Ġinstance ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġflag Shard No More InProgress ( shard Id : ĠString , Ġhandler : ĠHandler < AsyncResult < Boolean >>) ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠPersistence Ġof Ġthe Ġcurrent Ġ" cursor " Ġor Ġin Ġcontext Ġof ĠK inesis Ġsequence Ġnumber Ġof Ġthe Ġlatest Ġsuccessful Ġpro ceeded ĊĠĠĠĠ Ġ* Ġrecord . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġsave Consumer Shard SequenceNumber ( ĊĠĠĠĠĠĠĠ Ġshard Id : ĠString , ĊĠĠĠĠĠĠĠ Ġsequence Number : ĠString , ĊĠĠĠĠĠĠĠ Ġiterator Position : ĠSequence Number Iterator Position , ĊĠĠĠĠĠĠĠ Ġhandler : ĠHandler < AsyncResult < Void ? >> ĊĠĠĠ Ġ) ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠShould Ġcall Ġ[ handler ] Ġwith Ġthe Ġmost Ġrecent Ġvalue Ġof Ġthe Ġ[ save Consumer Shard SequenceNumber ] Ġcall . ĊĠĠĠĠ Ġ* ĠThe Ġ[ JsonObject ] Ġparameter Ġof Ġ[ handler ] Ġrepresents Ġa Ġ[ SequenceNumber ] Ġinstance . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġget Consumer Shard SequenceNumber ( shard Id : ĠString , Ġhandler : ĠHandler < AsyncResult < JsonObject ? >>) ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠDe letion Ġof Ġany Ġvalue Ġof Ġprevious Ġ[ save Consumer Shard SequenceNumber ] Ġcalls . Ġ[ get Consumer Shard SequenceNumber ] Ġshould Ġcall ĊĠĠĠĠ Ġ* Ġits Ġhandler Ġwith Ġnull . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġdelete Shard SequenceNumber ( shard Id : ĠString , Ġhandler : ĠHandler < AsyncResult < Boolean >>) ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠS aves Ġa Ġfinished Ġflag Ġfor Ġthe Ġgiven Ġshard Ġid . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġsave Finished Shard ( shard Id : ĠString , Ġexpiration Millis : ĠLong , Ġhandler : ĠHandler < AsyncResult < Void ? >>) ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠCalls Ġ[ handler ] Ġwith Ġa Ġlist Ġof Ġfinished Ġshard Ġids . ĠBy Ġprevious Ġcalls Ġon Ġ[ save Finished Shard ]) ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġget Finished Shard Ids ( handler : ĠHandler < AsyncResult < List < String >> >) ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠShould Ġflag Ġthe Ġ[ parent Shard Id ] Ġof Ġ[ child Shard Id ] Ġas Ġfinished Ġand Ġreturn Ġamount Ġof Ġfinished Ġparent Ġsh ards . ĊĠĠĠĠ Ġ* ĠThis Ġcan Ġbe Ġcalled Ġmultiple Ġtimes Ġfor Ġthe Ġsame Ġparent . ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĠAs Ġmerge Ġparents Ġare Ġconsumed Ġfrom Ġdifferent ĠVK CO Ġinstances , Ġthe Ġtimestamp Ġwhere Ġeach Ġof Ġthem Ġwould Ġbe Ġfinished ĊĠĠĠĠ Ġ* Ġwould Ġbe Ġdiffer . ĠTo Ġdetermine Ġif Ġa Ġchild Ġshard Ġcould Ġconsumed Ġas Ġboth Ġparent ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġflag Merge Parent Res h arding Ready ( parent Shard Id : ĠString , Ġchild Shard Id : ĠString , Ġhandler : ĠHandler < AsyncResult < Boolean >>) ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠDeletes Ġthe Ġincrement ation Ġstate Ġof Ġ[ flag Merge Parent Res h arding Ready ]. ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġfun Ġdelete Merge Parents Res h arding Ready Flag ( child Shard Id : ĠString , Ġhandler : ĠHandler < AsyncResult < Int >>) Ċ } Ċ Ċ class ĠSh ard State Persistence Service Async ( private Ġval Ġdelegate : ĠSh ard State Persistence Service ) Ġ: ĊĠĠĠ ĠSh ard State Persistence Service Ġby Ġdelegate Ġ{ ĊĠĠĠ Ġsuspend Ġfun Ġget Shard Ids InProgress (): ĠSh ard IdList Ġ= ĊĠĠĠĠĠĠĠ Ġawait Result < List < String >> Ġ{ Ġget Shard Ids InProgress ( it ) Ġ}. map Ġ{ Ġit . as Shard Id Typed () Ġ} ĊĊĠĠĠ Ġsuspend Ġfun Ġflag Shard InProgress ( shard Id : ĠSh ard Id ) Ġ= Ġawait Result < Boolean > Ġ{ Ġflag Shard InProgress ( shard Id . id , Ġit ) Ġ} ĊĊĠĠĠ Ġsuspend Ġfun Ġflag Shard No More InProgress ( shard Id : ĠSh ard Id ) Ġ= ĊĠĠĠĠĠĠĠ Ġawait Result < Boolean > Ġ{ Ġflag Shard No More InProgress ( shard Id . id , Ġit ) Ġ} ĊĊĠĠĠ Ġsuspend Ġfun Ġsave Consumer Shard SequenceNumber ( shard Id : ĠSh ard Id , Ġsequence Number : ĠSequence Number ) Ġ= ĊĠĠĠĠĠĠĠ Ġawait Result < Void ?> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsave Consumer Shard SequenceNumber ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġshard Id . id , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsequence Number . number , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsequence Number . iterator Position , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġsuspend Ġfun Ġget Consumer Shard SequenceNumber ( shard Id : ĠSh ard Id ): ĠSequence Number ? Ġ= ĊĠĠĠĠĠĠĠ Ġawait Result < JsonObject ?> Ġ{ Ġget Consumer Shard SequenceNumber ("$ shard Id ", Ġit ) Ġ} ?. map To ( SequenceNumber :: class . java ) ĊĊĠĠĠ Ġsuspend Ġfun Ġdelete Shard SequenceNumber ( shard Id : ĠSh ard Id ) Ġ= ĊĠĠĠĠĠĠĠ Ġawait Result < Boolean > Ġ{ Ġdelete Shard SequenceNumber ("$ shard Id ", Ġit ) Ġ} ĊĊĠĠĠ Ġsuspend Ġfun Ġsave Finished Shard ( shard Id : ĠSh ard Id , Ġexpiration Millis : ĠLong ) Ġ= ĊĠĠĠĠĠĠĠ Ġawait Result < Void ?> Ġ{ Ġsave Finished Shard ("$ shard Id ", Ġexpiration Millis , Ġit ) Ġ} ĊĊĠĠĠ Ġsuspend Ġfun Ġget Finished Shard Ids (): ĠSh ard IdList Ġ= ĊĠĠĠĠĠĠĠ Ġawait Result < List < String >> Ġ{ Ġget Finished Shard Ids ( it ) Ġ}. map Ġ{ Ġit . as Shard Id Typed () Ġ} ĊĊĠĠĠ Ġsuspend Ġfun Ġflag Merge Parent Ready To Res hard ( parent Shard Id : ĠSh ard Id , Ġchild Shard Id : ĠSh ard Id ) Ġ= ĊĠĠĠĠĠĠĠ Ġawait Result < Boolean > Ġ{ Ġthis . flag Merge Parent Res h arding Ready ("$ parent Shard Id ", Ġ"$ child Shard Id ", Ġit ) Ġ} ĊĊĠĠĠ Ġsuspend Ġfun Ġdelete Merge Parents Res h arding Ready Flag ( child Shard Id : ĠSh ard Id ) Ġ= ĊĠĠĠĠĠĠĠ Ġawait Result < Int > Ġ{ Ġdelete Merge Parents Res h arding Ready Flag ( child Shard Id . id , Ġit ) Ġ} Ċ } Ċ