/* Ċ Ġ* ĠSPDX - License - Identifier : ĠApache - 2 . 0 Ċ Ġ* Ċ Ġ* ĠThe ĠOpen Search ĠContributors Ġrequire Ġcontributions Ġmade Ġto Ċ Ġ* Ġthis Ġfile Ġbe Ġlicensed Ġunder Ġthe ĠApache - 2 . 0 Ġlicense Ġor Ġa Ċ Ġ* Ġcompatible Ġopen Ġsource Ġlicense . Ċ Ġ* Ċ Ġ* ĠMod ifications ĠCopyright ĠOpen Search ĠContributors . ĠSee Ċ Ġ* ĠGitHub Ġhistory Ġfor Ġdetails . Ċ Ġ*/ Ċ Ċ package Ġorg . opense arch . replication . task . shard Ċ Ċ import Ġorg . opense arch . replication . Replication Settings Ċ import Ġorg . opense arch . replication . action . changes . Get Changes Action Ċ import Ġorg . opense arch . replication . action . changes . Get Changes Request Ċ import Ġorg . opense arch . replication . action . changes . Get Changes Response Ċ import Ġorg . opense arch . replication . metadata . Replication Metadata Manager Ċ import Ġorg . opense arch . replication . metadata . Replication Overall State Ċ import Ġorg . opense arch . replication . metadata . state . RE PLICATION _ LAST _ KNOWN _ OVER ALL _ STATE Ċ import Ġorg . opense arch . replication . metadata . state . get Replication State Params For Index Ċ import Ġorg . opense arch . replication . seq no . Remote Cluster Retention Lease Helper Ċ import Ġorg . opense arch . replication . task . Cross Cluster Replication Task Ċ import Ġorg . opense arch . replication . task . Replication State Ċ import Ġorg . opense arch . replication . util . indices Service Ċ import Ġorg . opense arch . replication . util . stack Trace ToString Ċ import Ġorg . opense arch . replication . util . suspend Execute With Retries Ċ import Ġkotlinx . coroutines . Cancellation Exception Ċ import Ġkotlinx . coroutines . Coroutine ExceptionHandler Ċ import Ġkotlinx . coroutines . Coroutine Scope Ċ import Ġkotlinx . coroutines . Non C ancellable Ċ import Ġkotlinx . coroutines . Obsolete Cor outines Api Ċ import Ġkotlinx . coroutines . coroutine Scope Ċ import Ġkotlinx . coroutines . delay Ċ import Ġkotlinx . coroutines . isActive Ċ import Ġkotlinx . coroutines . launch Ċ import Ġkotlinx . coroutines . supervisor Scope Ċ import Ġkotlinx . coroutines . sync . Semaphore Ċ import Ġkotlinx . coroutines . with Context Ċ import Ġorg . opense arch . Open Search Exception Ċ import Ġorg . opense arch . Open Search TimeoutException Ċ import Ġorg . opense arch . client . Client Ċ import Ġorg . opense arch . cluster . Cluster Changed Event Ċ import Ġorg . opense arch . cluster . Cluster State Listener Ċ import Ġorg . opense arch . cluster . service . Cluster Service Ċ import Ġorg . opense arch . common . logging . Log gers Ċ import Ġorg . opense arch . index . seq no . Retention Lease Actions Ċ import Ġorg . opense arch . index . seq no . Retention Lease Invalid Ret aining Seq No Exception Ċ import Ġorg . opense arch . index . seq no . Retention Lease NotFoundException Ċ import Ġorg . opense arch . index . shard . Shard Id Ċ import Ġorg . opense arch . persistent . Persistent Task State Ċ import Ġorg . opense arch . persistent . Persistent Tasks Node Service Ċ import Ġorg . opense arch . rest . Rest Status Ċ import Ġorg . opense arch . tasks . TaskId Ċ import Ġorg . opense arch . thread pool . ThreadPool Ċ import Ġorg . opense arch . transport . Node Not Connected Exception Ċ import Ġjava . time . Duration ĊĊ Ċ class ĠSh ard Replication Task ( id : ĠLong , Ġtype : ĠString , Ġaction : ĠString , Ġdescription : ĠString , Ġparent Task : ĠTask Id , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġparams : ĠSh ard Replication Params , Ġexecutor : ĠString , Ġcluster Service : ĠCluster Service , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthread Pool : ĠThreadPool , Ġclient : ĠClient , Ġreplication Metadata Manager : ĠRep lication Metadata Manager , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreplication Settings : ĠRep lication Settings , Ġprivate Ġval Ġfollow er Cluster Stats : ĠFollow er Cluster Stats ) ĊĠĠĠ Ġ: ĠCross Cluster Replication Task ( id , Ġtype , Ġaction , Ġdescription , Ġparent Task , Ġempty Map (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġexecutor , Ġcluster Service , Ġthread Pool , Ġclient , Ġreplication Metadata Manager , Ġreplication Settings ) Ġ{ ĊĊĠĠĠ Ġoverride Ġval Ġleader Alias : ĠString Ġ= Ġparams . leader Alias ĊĠĠĠ Ġoverride Ġval Ġfollow er Index Name : ĠString Ġ= Ġparams . follow er Shard Id . index Name ĊĠĠĠ Ġprivate Ġval Ġleader Shard Id Ġ= Ġparams . leader Shard Id ĊĠĠĠ Ġprivate Ġval Ġfollow er Shard Id Ġ= Ġparams . follow er Shard Id ĊĠĠĠ Ġprivate Ġval Ġremote Client Ġ= Ġclient . get Remote Cluster Client ( leader Alias ) ĊĠĠĠ Ġprivate Ġval Ġretention Lease Helper Ġ= ĠRemote Cluster Retention Lease Helper ( cluster Service . cluster Name . value (), Ġremote Client ) ĊĠĠĠ Ġprivate Ġvar Ġpaused Ġ= Ġfalse ĊĠĠĠ Ġprivate Ġvar Ġlast Lease Renew al Millis Ġ= ĠSystem . currentTimeMillis () ĊĊĠĠĠ Ġ// Start Ġback Off Ġfor Ġexceptions Ġwith Ġa Ġsecond ĊĠĠĠ Ġprivate Ġval Ġinitial Backoff Millis Ġ= Ġ 1 0 0 0 L ĊĠĠĠ Ġ// Start Ġback Off Ġfor Ġexceptions Ġwith Ġa Ġsecond ĊĠĠĠ Ġprivate Ġvar Ġback Off For Retry Ġ= Ġinitial Backoff Millis ĊĠĠĠ Ġ// Max Ġtimeout Ġfor Ġbackoff ĊĠĠĠ Ġprivate Ġval Ġmax TimeOut Ġ= Ġ 6 0 0 0 0 L ĊĠĠĠ Ġ// Backoff Ġfactor Ġafter Ġevery Ġretry ĊĠĠĠ Ġprivate Ġval Ġfactor Ġ= Ġ 2 . 0 ĊĊĠĠĠ Ġprivate Ġval Ġcluster State Listener For Task Inter ruption Ġ= ĠCluster State Listener For Task Inter ruption () ĊĊĠĠĠ Ġoverride Ġval Ġlog Ġ= ĠLog gers . getLogger ( java Class , Ġfollow er Shard Id ) !! ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġfun ĠtaskId For Shard ( shard Id : ĠSh ard Id ) Ġ= Ġ" replication :${ shard Id }" ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Obsolete Cor outines Api ĊĠĠĠ Ġoverride Ġsuspend Ġfun Ġexecute ( scope : ĠCor outine Scope , ĠinitialState : ĠPersistent Task State ?) Ġ{ ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠThe ĠCor outine ExceptionHandler Ġinstalled Ġis Ġmainly Ġused Ġto Ġcatch Ġthe Ġexception Ġfrom Ġreplication Ġreplay ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġlogic Ġand Ġpersist Ġthe Ġfailure Ġreason . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġdownstream Exception : ĠThrowable ? Ġ= Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġhandler Ġ= ĠCor outine ExceptionHandler Ġ{ Ġ_, Ġexception Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog Error (" Shard Replication Task : ĠCa ught Ġdownstream Ġexception Ġ${ exception . stack Trace ToString () }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdownstream Exception Ġ= Ġexception ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWrap Ġthe Ġactual Ġreplication Ġreplay Ġlogic Ġin ĠSuper visor Coroutine Ġand Ġan Ġinner Ġcoroutine Ġso Ġthat Ġwe Ġhave ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġbetter Ġcontrol Ġover Ġexception Ġpropagation . ĠEss entially Ġany Ġfailures Ġfrom Ġinner Ġreplication Ġlogic Ġwill ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġnot Ġcancel Ġthe Ġparent Ġcoroutine Ġand Ġthe Ġexception Ġis Ġcaught Ġby Ġthe Ġinstalled ĠCor outine ExceptionHandler ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠThe Ġonly Ġpath Ġfor Ġcancellation Ġof Ġthis Ġouter Ġcoroutine Ġis Ġexternal Ġexplicit Ġcancellation Ġ( pause Ġlogic , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġtask Ġbeing Ġcancelled Ġby ĠAPI Ġetc ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠCheckout Ġout Ġthe Ġfollowing Ġfor Ġdetails ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġhttps :// kotlin lang . org / docs / exception - handling . html # super vision - scope ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsupervisor Scope Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlaunch ( handler ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġreplicate ( this ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠNon - null Ġdownstream Exception Ġimplies , Ġexception Ġin Ġinner Ġreplication Ġcode . ĠIn Ġsuch Ġcases Ġwe Ġmark Ġand ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġcapture Ġthe ĠFailed State Ġand Ġwait Ġfor Ġparent ĠIndex Replication Task Ġto Ġtake Ġaction . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠNote Ġthat Ġwe Ġdon 't Ġtake Ġthe Ġaction Ġto Ġpause / stop Ġdirectly Ġfrom Ġthis ĠSh ard Replication Task Ġsince ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠIndex Replication Task Ġcan Ġchoose Ġthe Ġappropriate Ġaction Ġbased Ġon Ġfailures Ġseen Ġfrom Ġmultiple Ġsh ards . ĠThis ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġapproach Ġalso Ġavoids Ġcontent ion Ġdue Ġto Ġconcurrency . ĠFinally Ġit Ġkeeps Ġthe Ġscope Ġof Ġresponsibility Ġof ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠSh ard Replication Task Ġto ĠSh ard Replication Task Ġalone . ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( null Ġ!= Ġdownstream Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠExplicit Ġcast Ġis Ġrequired Ġfor Ġchanging Ġclo s ures ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġthrowable : ĠThrowable Ġ= Ġdownstream Exception Ġas ĠThrowable ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwith Context ( Non C ancellable ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog Info (" Go ing Ġto Ġmark ĠSh ard Replication Task Ġas ĠFailed Ġwith Ġ${ throwable . stack Trace ToString () }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġupdate Task State ( Failed State ( to ES Exception ( throwable ))) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( inner : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog Info (" Encounter ed Ġexception Ġwhile Ġtrying Ġto Ġpersist Ġfailed Ġstate Ġ${ inner . stack Trace ToString () }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠWe Ġare Ġnot Ġpropag ating Ġfailure Ġhere Ġand Ġwill Ġlet Ġthe Ġshard Ġtask Ġbe Ġfailed Ġafter Ġwaiting . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠAfter Ġmarking ĠFailed State , ĠIndex Replication Task Ġwill Ġaction Ġon Ġit Ġby Ġpa using Ġor Ġstopping Ġall Ġshard ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġreplication Ġtasks . ĠThis ĠSh ard Replication Task Ġshould Ġalso Ġthus Ġreceive Ġthe Ġpause / stop Ġvia ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// Ġcancellation . ĠWe Ġthus Ġwait Ġfor Ġwait Millis Ġduration . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġwait Millis Ġ= ĠDuration . of Minutes ( 1 0 ). to Millis () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog Info (" Waiting Ġ$ wait Millis Ġmillis Ġfor ĠIndex Replication Task Ġto Ġrespond Ġto Ġfailure Ġof Ġshard Ġtask ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( wait Millis ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠIf Ġnothing Ġhappened , Ġwe Ġpropagate Ġexception Ġand Ġmark Ġthe Ġtask Ġas Ġfailed . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġthrowable ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠC ancellation Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠNothing Ġto Ġdo Ġhere Ġand Ġwe Ġdon 't Ġpropagate Ġcancellation Ġexception Ġfurther ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog Info (" Received Ġcancellation Ġof ĠSh ard Replication Task Ġ${ e . stack Trace ToString () }") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġto ES Exception ( t : ĠThrowable ?): ĠOpen Search Exception Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( t Ġis ĠOpen Search Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġt ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġmsg Ġ= Ġt ?. message Ġ?: Ġt ?. java Class ?. name Ġ?: Ġ" Unknown Ġfailure Ġencountered " ĊĠĠĠĠĠĠĠ Ġreturn ĠOpen Search Exception ( msg , Ġt ) ĊĠĠĠ Ġ} ĊĊĊĠĠĠ Ġoverride Ġsuspend Ġfun Ġcleanup () Ġ{ ĊĠĠĠĠĠĠĠ Ġ/* ĠThis Ġis Ġto Ġminim ise Ġoverhead Ġof Ġcalling Ġan Ġadditional Ġlistener Ġas ĊĠĠĠĠĠĠĠ Ġ* Ġit Ġcontinues Ġto Ġbe Ġcalled Ġeven Ġafter Ġthe Ġtask Ġis Ġcompleted . ĊĠĠĠĠĠĠĠĠ Ġ*/ ĊĠĠĠĠĠĠĠ Ġcluster Service . remove Listener ( cluster State Listener For Task Inter ruption ) ĊĠĠĠĠĠĠĠ Ġthis . follow er Cluster Stats . stats . remove ( follow er Shard Id ) ĊĠĠĠĠĠĠĠ Ġif Ġ( paused ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog Debug (" Pa using Ġand Ġnot Ġremoving Ġlease Ġfor Ġindex Ġ$ follow er Index Name Ġand Ġshard Ġ$ follow er Shard Id Ġtask ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġretention Lease Helper . attempt Retention Lease Removal ( leader Shard Id , Ġfollow er Shard Id ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġadd Listener To Interrupt Task () Ġ{ ĊĠĠĠĠĠĠĠ Ġcluster Service . addListener ( cluster State Listener For Task Inter ruption ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġinner Ġclass ĠCluster State Listener For Task Inter ruption Ġ: ĠCluster State Listener Ġ{ ĊĠĠĠĠĠĠĠ Ġoverride Ġfun Ġcluster Changed ( event : ĠCluster Changed Event ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog Debug (" Cluster Ġmetadata Ġlistener Ġinvoked Ġon Ġshard Ġtask ...") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( event . metadata Changed ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġreplication State Params Ġ= Ġget Replication State Params For Index ( cluster Service , Ġfollow er Shard Id . index Name ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( replication State Params Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( Persistent Tasks Node Service . Status ( State . STARTED ) Ġ== Ġstatus ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcancel Task (" Shard Ġreplication Ġtask Ġreceived Ġan Ġinterrupt .") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġif Ġ( replication State Params [ RE PLICATION _ LAST _ KNOWN _ OVER ALL _ STATE ] Ġ== ĠRep lication Overall State . PA USED . name ){ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog Info (" Pause Ġstate Ġreceived Ġfor Ġindex Ġ$ follow er Index Name . ĠC ancell ing Ġ$ follow er Shard Id Ġtask ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpaused Ġ= Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcancel Task (" Shard Ġreplication Ġtask Ġreceived Ġpause .") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġindices Or Shards () Ġ= ĠlistOf ( follow er Shard Id ) ĊĊĠĠĠ Ġ@ Obsolete Cor outines Api ĊĠĠĠ Ġprivate Ġsuspend Ġfun Ġreplicate ( scope : ĠCor outine Scope ) Ġ{ ĊĠĠĠĠĠĠĠ Ġupdate Task State ( Following State ) ĊĠĠĠĠĠĠĠ Ġval Ġfollow er Index Service Ġ= Ġindices Service . index Service Safe ( follow er Shard Id . index ) ĊĠĠĠĠĠĠĠ Ġval Ġindex Shard Ġ= Ġfollow er Index Service . get Shard ( follow er Shard Id . id ) ĊĠĠĠĠĠĠĠ Ġ// ĠAdding Ġretention Ġlease Ġat Ġlast Sync ed Global Checkpoint . ĠThis Ġmakes Ġsure ĊĠĠĠĠĠĠĠ Ġ// Ġnew Ġtasks Ġspawn ed Ġafter Ġnode Ġchanges / shard Ġmov ements Ġare Ġhandled Ġproperly ĊĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog Info (" Adding Ġretention Ġlease ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġretention Lease Helper . add Retention Lease ( leader Shard Id , Ġindex Shard . last Sync ed Global Checkpoint , Ġfollow er Shard Id ) ĊĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠRet ention Lease Invalid Ret aining Seq No Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog Info (" Retention Ġlease Ġadd Ġfailed . ĠIgn oring Ġ${ e . stack Trace ToString () }") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġadd Listener To Interrupt Task () ĊĠĠĠĠĠĠĠ Ġthis . follow er Cluster Stats . stats [ follow er Shard Id ] Ġ= ĠFollow er Shard Metric () ĊĊĠĠĠĠĠĠĠ Ġ// ĠSince Ġthis Ġsetting Ġis Ġnot Ġdynamic , Ġsetting Ġupdate Ġwould Ġonly Ġreflect Ġafter Ġpause - resume Ġor Ġon Ġa Ġnew Ġreplication Ġjob . ĊĠĠĠĠĠĠĠ Ġval Ġrate Limiter Ġ= ĠSem aphore ( replication Settings . readers Per Shard ) ĊĠĠĠĠĠĠĠ Ġval Ġsequ encer Ġ= ĠTrans log Se qu encer ( scope , Ġreplication Metadata , Ġfollow er Shard Id , Ġleader Alias , Ġleader Shard Id . index Name , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠTask Id ( cluster Service . nodeName , Ġid ), Ġclient , Ġindex Shard . local Checkpoint , Ġfollow er Cluster Stats ) ĊĊĠĠĠĠĠĠĠ Ġval Ġchange Tracker Ġ= ĠSh ard Replication Changes Tracker ( index Shard , Ġreplication Settings ) ĊĠĠĠĠĠĠĠ Ġcoroutine Scope Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( isActive ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrate Limiter . acquire () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlaunch Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog Debug (" Sp aw ning Ġthe Ġreader ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġbatch To Fetch Ġ= Ġchange Tracker . request Batch To Fetch () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġfrom Seq No Ġ= Ġbatch To Fetch . first ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġto Seq No Ġ= Ġbatch To Fetch . second ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog Debug (" Getting Ġchanges Ġ$ from Seq No -$ to Seq No ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġchanges Response Ġ= Ġget Changes ( from Seq No , Ġto Seq No ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog Info (" Got Ġ${ changes Response . changes . size } Ġchanges Ġstarting Ġfrom Ġseq No : Ġ$ from Seq No ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġsequ encer . send ( changes Response ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog Debug (" pus hed Ġto Ġsequ encer Ġ$ from Seq No -$ to Seq No ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġchange Tracker . update Batch F etched ( true , Ġfrom Seq No , Ġto Seq No , Ġchanges Response . changes . last OrNull ()?. seq No () Ġ?: Ġfrom Seq No Ġ- Ġ 1 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġchanges Response . last Sync ed Global Checkpoint ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// reset Ġbackoff Ġafter Ġevery Ġsuccessful Ġget Changes Ġcall ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġback Off For Retry Ġ= Ġinitial Backoff Millis ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠOpen Search TimeoutException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// TimeoutException Ġis Ġthrown Ġif Ġleader Ġfails Ġto Ġsend Ġnew Ġchanges Ġin Ġ 1 Ġminute , Ġso Ġwe Ġdont Ġneed Ġa Ġbackoff Ġagain Ġhere Ġfor Ġthis Ġexception ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog Info (" Timed Ġout Ġwaiting Ġfor Ġnew Ġchanges . ĠCurrent Ġseq No : Ġ$ from Seq No . Ġ$ e ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġchange Tracker . update Batch F etched ( false , Ġfrom Seq No , Ġto Seq No , Ġfrom Seq No Ġ- Ġ 1 ,- 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠNode Not Connected Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfollow er Cluster Stats . stats [ follow er Shard Id ] !!. ops Read Failures . add AndGet ( 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog Info (" Node Ġnot Ġconnected . ĠRetry ing Ġrequest Ġusing Ġa Ġdifferent Ġnode . Ġ${ e . stack Trace ToString () }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( back Off For Retry ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġback Off For Retry Ġ= Ġ( back Off For Retry Ġ* Ġfactor ). to Long (). coerce At Most ( max TimeOut ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġchange Tracker . update Batch F etched ( false , Ġfrom Seq No , Ġto Seq No , Ġfrom Seq No Ġ- Ġ 1 ,- 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( e : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfollow er Cluster Stats . stats [ follow er Shard Id ] !!. ops Read Failures . add AndGet ( 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog Info (" Unable Ġto Ġget Ġchanges Ġfrom Ġseq No : Ġ$ from Seq No . Ġ${ e . stack Trace ToString () }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġchange Tracker . update Batch F etched ( false , Ġfrom Seq No , Ġto Seq No , Ġfrom Seq No Ġ- Ġ 1 ,- 1 ) ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// ĠPro pag ate Ġ 4 xx Ġexceptions Ġup Ġthe Ġchain Ġand Ġhalt Ġreplication Ġas Ġthey Ġare Ġir recover able ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġrange 4 xx Ġ= Ġ 4 0 0 . range To ( 4 9 9 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( e Ġis ĠOpen Search Exception Ġ&& ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrange 4 xx . contains ( e . status (). status ) Ġ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( e . status (). status Ġ== ĠRest Status . TOO _ MAN Y _ REQUEST S . status ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfollow er Cluster Stats . stats [ follow er Shard Id ] !!. ops Read Th rott les . add AndGet ( 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġe ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdelay ( back Off For Retry ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġback Off For Retry Ġ= Ġ( back Off For Retry Ġ* Ġfactor ). to Long (). coerce At Most ( max TimeOut ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġfinally Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrate Limiter . release () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// renew Ġretention Ġlease Ġwith Ġglobal Ġcheckpoint Ġso Ġthat Ġany Ġshard Ġthat Ġp icks Ġup Ġshard Ġreplication Ġtask Ġhas Ġdata Ġuntil Ġthen . ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġretention Lease Helper . renew Retention Lease ( leader Shard Id , Ġindex Shard . last Sync ed Global Checkpoint , Ġfollow er Shard Id ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfollow er Cluster Stats . stats [ follow er Shard Id ] !!. follow er Checkpoint Ġ= Ġindex Shard . last Sync ed Global Checkpoint ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlast Lease Renew al Millis Ġ= ĠSystem . currentTimeMillis () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġcatch Ġ( ex : ĠException ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( ex ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠRet ention Lease NotFoundException Ġ-> Ġ Ġthrow Ġex ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠRet ention Lease Invalid Ret aining Seq No Exception Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( System . currentTimeMillis () Ġ- Ġlast Lease Renew al Millis Ġ> Ġ Ġreplication Settings . lease Renew al Max Failure Duration . millis ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . error (" Retention Ġlease Ġrenew al Ġhas Ġbeen Ġfailing Ġfor Ġlast Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ"${ replication Settings . lease Renew al Max Failure Duration . minutes } Ġminutes ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġex ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . error (" Retention Ġlease Ġrenew al Ġfailed . ĠIgn oring . Ġ${ ex . message }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġlog Info (" Exception Ġrenew ing Ġretention Ġlease . ĠNot Ġan Ġissue Ġ- Ġ${ ex . stack Trace ToString () }") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġsequ encer . close () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġsuspend Ġfun Ġget Changes ( from Seq No : ĠLong , Ġto Seq No : ĠLong ): ĠGet Changes Response Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġremote Client Ġ= Ġclient . get Remote Cluster Client ( leader Alias ) ĊĠĠĠĠĠĠĠ Ġval Ġrequest Ġ= ĠGet Changes Request ( leader Shard Id , Ġfrom Seq No , Ġto Seq No ) ĊĠĠĠĠĠĠĠ Ġvar Ġchanges Resp Ġ= Ġ Ġremote Client . suspend Execute With Retries ( replication Metadata Ġ= Ġreplication Metadata , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġaction Ġ= ĠGet Changes Action . INSTANCE , Ġreq Ġ= Ġrequest , Ġlog Ġ= Ġlog ) ĊĠĠĠĠĠĠĠ Ġfollow er Cluster Stats . stats [ follow er Shard Id ] !!. leader Checkpoint Ġ= Ġchanges Resp . last Sync ed Global Checkpoint ĊĠĠĠĠĠĠĠ Ġfollow er Cluster Stats . stats [ follow er Shard Id ] !!. ops Read . add AndGet ( changes Resp . changes . size . to Long ()) ĊĠĠĠĠĠĠĠ Ġreturn Ġchanges Resp ĊĠĠĠ Ġ} ĊĠĠĠ Ġprivate Ġfun Ġlog Debug ( msg : ĠString ) Ġ{ ĊĠĠĠĠĠĠĠ Ġlog . debug ("${ Thread . currentThread (). name }: Ġ$ msg ") ĊĠĠĠ Ġ} ĊĠĠĠ Ġprivate Ġfun Ġlog Info ( msg : ĠString ) Ġ{ ĊĠĠĠĠĠĠĠ Ġlog . info ("${ Thread . currentThread (). name }: Ġ$ msg ") ĊĠĠĠ Ġ} ĊĠĠĠ Ġprivate Ġfun Ġlog Error ( msg : ĠString ) Ġ{ ĊĠĠĠĠĠĠĠ Ġlog . error ("${ Thread . currentThread (). name }: Ġ$ msg ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun ĠtoString (): ĠString Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġ" Shard Replication Task ( from =${ leader Alias }$ leader Shard Id Ġto =$ follow er Shard Id )" ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġreplication Task Response (): ĠCross Cluster Replication Task Response Ġ{ ĊĠĠĠĠĠĠĠ Ġ// ĠC ancellation Ġand Ġvalid Ġexecution s Ġare Ġmarked Ġas Ġcompleted ĊĠĠĠĠĠĠĠ Ġreturn ĠCross Cluster Replication Task Response ( Replication State . COMPLETED . name ) ĊĠĠĠ Ġ} Ċ } Ċ