package Ġcom . te amb ition . kafka . connect . mongo . database Ċ Ċ import Ġcom . mongodb . Basic DB Object Ċ import Ġcom . mongodb . client . Mongo Collection Ċ import Ġcom . mongodb . client . model . Filters Ċ import Ġcom . te amb ition . kafka . connect . mongo . source . Mongo Source Offset Ċ import Ġorg . bson . B son Timestamp Ċ import Ġorg . bson . Document Ċ import Ġorg . bson . types . ObjectId Ċ import Ġorg . slf 4 j . LoggerFactory Ċ import Ġjava . util .* Ċ import Ġjava . util . concurrent . Concurrent Linked Queue Ċ Ċ /** Ċ Ġ* Ġ@ author ĠX u ĠJ ing xin Ċ Ġ*/ Ċ class ĠExport Reader ( ĊĠĠĠ Ġval Ġuri : ĠString , ĊĠĠĠ Ġval Ġdb : ĠString , ĊĠĠĠ Ġval Ġstart : ĠMongo Source Offset , ĊĠĠĠ Ġprivate Ġval Ġmessages : ĠConcurrent Linked Queue < Document >, ĊĠĠĠ Ġprivate Ġval Ġadditional Filter : ĠBasic DB Object ? Ġ= Ġnull Ċ ) Ġ{ ĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġlog Ġ= ĠLoggerFactory . getLogger ( Export Reader :: class . java ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġval Ġbatch Size Ġ= Ġ 1 0 0 ĊĠĠĠ Ġprivate Ġval Ġmongo Client Ġ= ĠMongo Client Loader . getClient ( uri ) ĊĠĠĠ Ġprivate Ġval Ġmax Message Size Ġ= Ġ 2 0 0 0 ĊĠĠĠ Ġvar Ġis Finished Ġ= Ġfalse ĊĊĠĠĠ Ġfun Ġrun () Ġ{ ĊĠĠĠĠĠĠĠ Ġvar Ġoffset Count Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġvar Ġoffset Id Ġ= Ġstart . objectId ĊĠĠĠĠĠĠĠ Ġval Ġmongo Collection Ġ= Ġget NS Collection ( db ) ĊĠĠĠĠĠĠĠ Ġlog . info (" Start Ġexport Ġreader Ġfor Ġdb Ġ{}, Ġstart Ġfrom Ġ{}", Ġdb , Ġstart ) ĊĠĠĠĠĠĠĠ Ġval Ġfilter Ġ= Ġadditional Filter ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ? . let Ġ{ Ġcombine Filter ( offset Id , Ġit ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ?: ĠFilters . gt ("_ id ", Ġoffset Id ) ĊĠĠĠĠĠĠĠ Ġmongo Collection ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. find () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. batch Size ( batch Size ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter ( filter ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. sort ( Document ("_ id ", Ġ 1 )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. as Sequence () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. forEach Ġ{ Ġdocument Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmessages . add ( format As Op Log ( document )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoffset Id Ġ= Ġdocument [" _ id "] Ġas ĠObjectId ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġoffset Count Ġ+= Ġ 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhile Ġ( messages . size Ġ> Ġmax Message Size ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠThread . sleep ( 1 0 0 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( offset Count Ġ% Ġ 1 0 0 0 Ġ== Ġ 0 L ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlog . info (" Export Ġprocessing Ġfor Ġdb Ġ{}, Ġcounts Ġ{}", Ġdb , Ġoffset Count ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġlog . info (" Export Ġfinished Ġfor Ġdb Ġ{}, Ġcount Ġ{}", Ġdb , Ġoffset Count ) ĊĠĠĠĠĠĠĠ Ġis Finished Ġ= Ġtrue ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġget NS Collection ( ns : ĠString ): ĠMongo Collection < Document > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġdb And Collection Ġ= Ġns . split ("\\ .". to Regex ()). drop Last While ( String :: isEmpty ) ĊĠĠĠĠĠĠĠ Ġval Ġns DB Ġ= Ġmongo Client . get Database ( db And Collection [ 0 ]) ĊĠĠĠĠĠĠĠ Ġreturn Ġns DB . get Collection ( db And Collection [ 1 ]) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġformat As Op Log ( doc : ĠDocument ): ĠDocument Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġop Doc Ġ= ĠDocument () ĊĠĠĠĠĠĠĠ Ġop Doc [" ts "] Ġ= ĠB son Timestamp (( Date (). time Ġ/ Ġ 1 0 0 0 ). toInt (), Ġ 0 ) ĊĠĠĠĠĠĠĠ Ġop Doc [" ns "] Ġ= Ġdb ĊĠĠĠĠĠĠĠ Ġop Doc [" initial Import "] Ġ= Ġtrue ĊĠĠĠĠĠĠĠ Ġop Doc [" op "] Ġ= Ġ" i " Ġ Ġ// ĠT reat Ġas Ġinsertion ĊĠĠĠĠĠĠĠ Ġop Doc [" o "] Ġ= Ġdoc ĊĠĠĠĠĠĠĠ Ġreturn Ġop Doc ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġcombine Filter ( id Offset : ĠObjectId , Ġadditional Filter : ĠBasic DB Object ) Ġ= ĊĠĠĠĠĠĠĠ ĠBasic DB Object ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmap Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ"\ $ and " Ġto ĠlistOf ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmap Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ"_ id " Ġto Ġmap Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ"\ $ gt " Ġto Ġid Offset ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġadditional Filter ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ) Ċ } Ċ