package Ġcom . te amb ition . kafka . connect . mongo . sink Ċ Ċ import Ġcom . te amb ition . kafka . connect . mongo . sink . Mongo Sink Config . Companion . DATABASE S _ CONFIG Ċ import Ġcom . te amb ition . kafka . connect . mongo . sink . Mongo Sink Config . Companion . MONGO _ URI _ CONFIG Ċ import Ġcom . te amb ition . kafka . connect . mongo . sink . Mongo Sink Config . Companion . SOURCE _ TOP ICS _ CONFIG Ċ import Ġorg . apache . kafka . common . config . Config Def Ċ import Ġorg . apache . kafka . common . utils . App Info Parser Ċ import Ġorg . apache . kafka . connect . connector . Task Ċ import Ġorg . apache . kafka . connect . errors . Connect Exception Ċ import Ġorg . apache . kafka . connect . sink . Sink Connector Ċ import Ġorg . apache . kafka . connect . util . Connector Utils Ċ import Ġorg . slf 4 j . LoggerFactory Ċ Ċ /** Ċ Ġ* Ġ@ author ĠX u ĠJ ing xin Ċ Ġ*/ Ċ class ĠMongo Sink Connector Ġ: ĠS ink Connector () Ġ{ ĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġlog Ġ= ĠLoggerFactory . getLogger ( Mongo Sink Connector :: class . java ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġvar Ġuri Ġ= Ġ"" ĊĠĠĠ Ġprivate Ġvar Ġtopics Ġ= Ġ"" ĊĠĠĠ Ġprivate Ġvar Ġdatabases Ġ= Ġ"" ĊĊĠĠĠ Ġoverride Ġfun Ġversion (): ĠString Ġ= ĠApp Info Parser . getVersion () ĊĊĠĠĠ Ġoverride Ġfun Ġtask Class (): ĠClass < out ĠTask > Ġ= ĠMongo Sink Task :: class . java ĊĊĠĠĠ Ġoverride Ġfun Ġconfig (): ĠConfig Def Ġ= ĠMongo Sink Config . config ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠGroup Ġtasks Ġby Ġnumber Ġof Ġtopics ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġoverride Ġfun Ġtask Configs ( max Tasks : ĠInt ): ĠMutable List < Mutable Map < String , ĠString >> Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġconfigs Ġ= Ġmutable ListOf < Mutable Map < String , ĠString >>() ĊĠĠĠĠĠĠĠ Ġval Ġtopics Ġ= Ġtopics . split ("," ). drop Last While ( String :: isEmpty ) ĊĠĠĠĠĠĠĠ Ġval Ġdatabases Ġ= Ġdatabases . split ("," ). drop Last While ( String :: isEmpty ) ĊĠĠĠĠĠĠĠ Ġval Ġnum Groups Ġ= ĠMath . min ( topics . size , Ġmax Tasks ) ĊĠĠĠĠĠĠĠ Ġval Ġtopics Group ed Ġ= ĠConnector Utils . group Partitions ( topics , Ġnum Groups ) ĊĠĠĠĠĠĠĠ Ġval Ġdb s Group ed Ġ= ĠConnector Utils . group Partitions ( databases , Ġnum Groups ) ĊĊĠĠĠĠĠĠĠ Ġfor Ġ( i Ġin Ġ 0 Ġuntil Ġnum Groups ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġconfig Ġ= Ġmutable Map Of < String , ĠString >() ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconfig [ MONGO _ URI _ CONFIG ] Ġ= Ġuri ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconfig [ SOURCE _ TOP ICS _ CONFIG ] Ġ= Ġtopics Group ed [ i ]. join ToString (",") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconfig [ DATABASE S _ CONFIG ] Ġ= Ġdb s Group ed [ i ]. join ToString (",") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconfigs . add ( config ) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġreturn Ġconfigs ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġstart ( props : ĠMap < String , ĠString >) Ġ{ ĊĠĠĠĠĠĠĠ Ġlog . trace (" Parsing Ġconfiguration : Ġ{}", Ġprops ) ĊĠĠĠĠĠĠĠ Ġuri Ġ= Ġget Required Prop ( props , ĠM ONGO _ URI _ CONFIG ) ĊĠĠĠĠĠĠĠ Ġtopics Ġ= Ġget Required Prop ( props , ĠSOURCE _ TOP ICS _ CONFIG ) ĊĠĠĠĠĠĠĠ Ġdatabases Ġ= Ġget Required Prop ( props , ĠDATABASE S _ CONFIG ) ĊĠĠĠĠĠĠĠ Ġval Ġtopics List Ġ= Ġtopics . split ("," ). drop Last While ( String :: isEmpty ) ĊĠĠĠĠĠĠĠ Ġval Ġdb s List Ġ= Ġdatabases . split ("," ). drop Last While ( String :: isEmpty ) ĊĠĠĠĠĠĠĠ Ġdb s List . forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġdb Collection Ġ= Ġit . split (".") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( db Collection . size Ġ!= Ġ 2 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠException (" Each Ġdatabase 's Ġpattern Ġshould Ġbe Ġdb . collection ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġif Ġ( topics List . size Ġ!= Ġdb s List . size ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠException (" Topics Ġand Ġdatabases Ġcount Ġshould Ġbe Ġthe Ġsame Ġin Ġsink Ġtask ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġlog . trace (" Configurations Ġ{}", Ġprops ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġstop () Ġ{ ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġget Required Prop ( props : ĠMap < String , ĠString >, Ġkey : ĠString ): ĠString Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġvalue Ġ= Ġprops [ key ] ĊĠĠĠĠĠĠĠ Ġif Ġ( value Ġ== Ġnull Ġ|| Ġvalue . isEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠConnect Exception (" Missing Ġ$ key Ġconfig ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġvalue ĊĠĠĠ Ġ} Ċ } Ċ