package Ġapp . cash . back fil a . client . mis k . internal Ċ Ċ import Ġapp . cash . back fil a . client . mis k . Back fill Ċ import Ġapp . cash . back fil a . client . mis k . embedded . Back fill Run Ċ import Ġapp . cash . back fil a . protos . client service . GetNext Batch Range Request Ċ import Ġapp . cash . back fil a . protos . client service . GetNext Batch Range Response Ċ import Ġapp . cash . back fil a . protos . client service . Key Range Ċ import Ġapp . cash . back fil a . protos . client service . Prepare Back fill Request Ċ import Ġapp . cash . back fil a . protos . client service . Prepare Back fill Response Ċ import Ġapp . cash . back fil a . protos . client service . Run Batch Request Ċ import Ġjava . util . Array Deque Ċ import Ġok io . ByteString Ċ import Ġok io . ByteString . Companion . encode Utf 8 Ċ Ċ /** Ċ Ġ* ĠA Ġsimple Ġback fill Ġrun Ġfor Ġtest Ġand Ġdevelopment . ĠUn like Ġthe Ġback fil a Ġservice , Ġthis Ġdoesn 't Ġuse Ċ Ġ* Ġthreads , Ġsleep ing , Ġor Ġbackoff . Ċ Ġ* Ċ Ġ* ĠThis Ġis Ġnot Ġthread Ġsafe . Ċ Ġ*/ Ċ internal Ġclass ĠEmbedded Back fill Run < B Ġ: ĠBack fill < *, Ġ* >>( ĊĠ Ġprivate Ġval Ġoperator : ĠBack fill Operator < *, Ġ* >, ĊĠ Ġoverride Ġval Ġdry Run : ĠBoolean , ĊĠ Ġoverride Ġval Ġparameters : ĠMap < String , ĠByte String >, ĊĠ Ġoverride Ġval Ġrange Start : ĠString ?, ĊĠ Ġoverride Ġval Ġrange End : ĠString ?, ĊĊĠ Ġoverride Ġvar Ġbatch Size : ĠLong Ġ= Ġ 1 0 0 L , ĊĠ Ġoverride Ġvar Ġscan Size : ĠLong Ġ= Ġ 1 0 _ 0 0 0 L , ĊĠ Ġoverride Ġvar Ġcompute Count Limit : ĠLong Ġ= Ġ 1 L Ċ ) Ġ: ĠBack fill Run < B > Ġ{ ĊĠ Ġoverride Ġval Ġback fill : ĠB ĊĠĠĠ Ġ// ĠThis Ġapi Ġon Ġthe Ġoperator Ġreturns Ġthe Ġsame Ġback fill Ġit Ġwas Ġinitialized Ġwith . ĊĠĠĠ Ġ@ Suppress (" UN CHECK ED _ CAST ") ĊĠĠĠ Ġget () Ġ= Ġoperator . back fill Ġas ĠB ĊĠ Ġoverride Ġval Ġprepare Back fill Response : ĠPrepare Back fill Response ĊĊĠ Ġprivate Ġval Ġpre compute Progress : ĠMap < String , ĠMutable Instance Cursor > ĊĠ Ġoverride Ġvar Ġpre compute Matching Count : ĠLong Ġ= Ġ 0 L ĊĠ Ġoverride Ġvar Ġpre compute Sc anned Count : ĠLong Ġ= Ġ 0 L ĊĊĠ Ġprivate Ġval Ġscan Progress : ĠMap < String , ĠMutable Instance Cursor > ĊĠ Ġoverride Ġval Ġinstance Progress Snapshot : ĠMap < String , ĠInstance Cursor > ĊĠĠĠ Ġget () Ġ= Ġscan Progress . map Values Ġ{ Ġit . value . snapshot () Ġ} ĊĊĠ Ġprivate Ġval Ġbatches To Run : ĠArray Deque < Batch Snapshot > Ġ= ĠArray Deque () ĊĠ Ġoverride Ġval Ġbatches To Run Snapshot : ĠList < Batch Snapshot > ĊĠĠĠ Ġget () Ġ= Ġbatches To Run . toList () ĊĊĠ Ġ/** ĠPrep ares Ġa Ġback fill Ġfor Ġa Ġrun . Ġ*/ ĊĠ Ġinit Ġ{ ĊĠĠĠ Ġprepare Back fill Response Ġ= Ġoperator . prepare Back fill ( Prepare Back fill Request . Builder () ĊĠĠĠĠĠĠĠ Ġ. back fill _ name ( operator . name ()) ĊĠĠĠĠĠĠĠ Ġ. range ( Key Range ( range Start ?. encode Utf 8 (), Ġrange End ?. encode Utf 8 ())) ĊĠĠĠĠĠĠĠ Ġ. parameters ( parameters ) ĊĠĠĠĠĠĠĠ Ġ. dry _ run ( dry Run ) ĊĠĠĠĠĠĠĠ Ġ. build ()) ĊĠĠĠ Ġpre compute Progress Ġ= Ġprepare Back fill Response . instances . associate Ġ{ ĊĠĠĠĠĠ Ġit . instance _ name Ġto ĠMutable Instance Cursor ( it . instance _ name , Ġit . back fill _ range ) ĊĠĠĠ Ġ} ĊĠĠĠ Ġscan Progress Ġ= Ġprepare Back fill Response . instances . associate Ġ{ ĊĠĠĠĠĠ Ġit . instance _ name Ġto ĠMutable Instance Cursor ( it . instance _ name , Ġit . back fill _ range ) ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġpre compute Scan (): ĠGet Next Batch Range Response Ġ{ ĊĠĠĠ Ġval Ġnot Done Ġ= Ġpre compute Progress . filter Not Ġ{ Ġit . value . done Ġ} ĊĠĠĠ Ġcheck ( not Done . isNotEmpty ()) Ġ{ Ġ" Nothing Ġleft Ġto Ġpre compute " Ġ} ĊĠĠĠ Ġval Ġcursor Ġ= Ġnot Done . values . first () ĊĠĠĠ Ġval Ġresponse Ġ= ĊĠĠĠĠĠĠĠ Ġoperator . getNext Batch Range ( GetNext Batch Range Request . Builder () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. instance _ name ( cursor . instance Name ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. back fill _ range ( cursor . key Range ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. previous _ end _ key ( cursor . previous End Key ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. parameters ( parameters ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. batch _ size ( batch Size ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. scan _ size ( scan Size ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. compute _ count _ limit ( compute Count Limit ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. pre computing ( true ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. dry _ run ( dry Run ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. build ()) ĊĠĠĠ Ġif Ġ( response . batches . isEmpty ()) Ġ{ ĊĠĠĠĠĠ Ġcursor . done Ġ= Ġtrue ĊĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠ Ġcursor . previous End Key Ġ= Ġresponse . batches . map Ġ{ Ġit . batch _ range . end Ġ}. max () ĊĠĠĠĠĠ Ġresponse . batches . on Each Ġ{ ĊĠĠĠĠĠĠĠ Ġpre compute Matching Count Ġ+= Ġit . matching _ record _ count ĊĠĠĠĠĠĠĠ Ġpre compute Sc anned Count Ġ+= Ġit . sc anned _ record _ count ĊĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠĠĠ Ġreturn Ġresponse ĊĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġpre compute Remaining () Ġ{ ĊĠĠĠ Ġdo Ġ{ ĊĠĠĠĠĠ Ġpre compute Scan () ĊĠĠĠ Ġ} Ġwhile Ġ(! finished Pre computing ()) ĊĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġfinished Pre computing () Ġ= Ġpre compute Progress . all Ġ{ Ġit . value . done Ġ} ĊĊĠ Ġoverride Ġfun Ġinstance Scan ( instance Name : ĠString ): ĠGet Next Batch Range Response Ġ{ ĊĠĠĠ Ġval Ġcursor Ġ= Ġscan Progress [ instance Name ] Ġ?: Ġerror ( ĊĠĠĠĠĠĠĠ Ġ" Instance Ġ$ instance Name Ġnot Ġfound . ĠValid Ġinstances Ġare Ġ${ scan Progress . keys }") ĊĠĠĠ Ġval Ġresponse Ġ= ĊĠĠĠĠĠĠĠ Ġoperator . getNext Batch Range ( GetNext Batch Range Request . Builder () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. instance _ name ( cursor . instance Name ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. back fill _ range ( cursor . key Range ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. previous _ end _ key ( cursor . previous End Key ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. parameters ( parameters ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. batch _ size ( batch Size ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. scan _ size ( scan Size ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. compute _ count _ limit ( compute Count Limit ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. dry _ run ( dry Run ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. build ()) ĊĠĠĠ Ġwhen Ġ( response . batches . isEmpty ()) Ġ{ ĊĠĠĠĠĠ Ġtrue Ġ-> Ġcursor . done Ġ= Ġtrue ĊĠĠĠĠĠ Ġfalse Ġ-> Ġcursor . previous End Key Ġ= Ġresponse . batches . map Ġ{ Ġit . batch _ range . end Ġ}. max () ĊĠĠĠ Ġ} ĊĠĠĠ Ġbatches To Run . addAll ( response . batches ĊĠĠĠĠĠĠĠ Ġ. filter Not Ġ{ Ġit . matching _ record _ count Ġ== Ġ 0 L Ġ} Ġ// ĠRemove Ġbatches Ġthat Ġhave Ġno Ġmatching Ġrecords . ĊĠĠĠĠĠĠĠ Ġ. map Ġ{ ĊĠĠĠĠĠĠĠĠĠ ĠBatch Snapshot ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġinstance Name , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . batch _ range , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . sc anned _ record _ count , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . matching _ record _ count ) ĊĠĠĠĠĠĠĠ Ġ}) ĊĠĠĠ Ġreturn Ġresponse ĊĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġsingle Scan (): ĠGet Next Batch Range Response Ġ{ ĊĠĠĠ Ġval Ġcursor Ġ= Ġscan Progress . values . first Ġ{ Ġ! it . done Ġ} ĊĠĠĠ Ġreturn Ġinstance Scan ( cursor . instance Name ) ĊĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġscan Remaining () Ġ{ ĊĠĠĠ Ġdo Ġ{ ĊĠĠĠĠĠ Ġsingle Scan () ĊĠĠĠ Ġ} Ġwhile Ġ(! finished Sc anning ()) ĊĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġfinished Sc anning () Ġ= Ġscan Progress . all Ġ{ Ġit . value . done Ġ} ĊĊĠ Ġoverride Ġfun Ġrun Batch () Ġ{ ĊĠĠĠ Ġcheck (! batches To Run . isEmpty ()) Ġ{ Ġ" There Ġmust Ġbe Ġbatches Ġto Ġrun " Ġ} ĊĠĠĠ Ġval Ġbatch Ġ= Ġbatches To Run . remove () ĊĠĠĠ Ġoperator . run Batch ( Run Batch Request . Builder () ĊĠĠĠĠĠĠĠ Ġ. instance _ name ( batch . instance Name ) ĊĠĠĠĠĠĠĠ Ġ. batch _ range ( batch . batch Range ) ĊĠĠĠĠĠĠĠ Ġ. parameters ( parameters ) ĊĠĠĠĠĠĠĠ Ġ. dry _ run ( dry Run ) ĊĠĠĠĠĠĠĠ Ġ. build ()) ĊĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġrun All Sc anned () Ġ{ ĊĠĠĠ Ġwhile Ġ( batches To Run . isNotEmpty ()) Ġ{ ĊĠĠĠĠĠ Ġrun Batch () ĊĠĠĠ Ġ} ĊĠ Ġ} ĊĊĠ Ġoverride Ġfun Ġcomplete () Ġ= Ġfinished Sc anning () Ġ&& Ġbatches To Run . isEmpty () ĊĊĠ Ġoverride Ġfun ĠtoString () Ġ= Ġ" Back fill Run [ ${ this . java Class . to Generic String ()} ]" Ċ } Ċ Ċ private Ġclass ĠMutable Instance Cursor ( ĊĠ Ġval Ġinstance Name : ĠString , ĊĠ Ġval Ġkey Range : ĠKey Range Ċ ) Ġ{ ĊĠ Ġvar Ġprevious End Key : ĠByte String ? Ġ= Ġnull ĊĠ Ġvar Ġdone : ĠBoolean Ġ= Ġfalse ĊĊĠ Ġfun Ġsnapshot () Ġ= ĠInstance Cursor ( ĊĠĠĠĠĠ Ġinstance Name , Ġkey Range , Ġprevious End Key , Ġdone ) Ċ } Ċ Ċ /** ĠImmutable Ġsnapshot Ġof Ġa Ġcursor . Ġ*/ Ċ data Ġclass ĠInstance Cursor ( ĊĠ Ġval Ġinstance Name : ĠString , ĊĠ Ġval Ġkey Range : ĠKey Range , ĊĠ Ġval Ġprevious End Key : ĠByte String ?, ĊĠ Ġval Ġdone : ĠBoolean Ċ ) Ġ{ ĊĠ Ġfun Ġutf 8 Range Start () Ġ= Ġkey Range . start ?. utf 8 () ĊĠ Ġfun Ġutf 8 Range End () Ġ= Ġkey Range . end ?. utf 8 () ĊĠ Ġfun Ġutf 8 Previous End Key () Ġ= Ġprevious End Key ?. utf 8 () Ċ } Ċ Ċ data Ġclass ĠBatch Snapshot ( ĊĠ Ġval Ġinstance Name : ĠString , ĊĠ Ġval Ġbatch Range : ĠKey Range , ĊĠ Ġval Ġscanned Record Count : ĠLong , ĊĠ Ġval Ġmatching Record Count : ĠLong Ċ ) Ġ{ ĊĠ Ġfun Ġutf 8 Range Start () Ġ= Ġbatch Range . start . utf 8 () ĊĠ Ġfun Ġutf 8 Range End () Ġ= Ġbatch Range . end . utf 8 () Ċ } Ċ