package Ġpipeline . kafka Ċ Ċ import Ġorg . apache . kafka . common . config . Config Def Ċ import Ġorg . apache . kafka . connect . connector . Task Ċ import Ġorg . apache . kafka . connect . data . Schema . STRING _ SCHEMA Ċ import Ġorg . apache . kafka . connect . source . Source Connector Ċ import Ġorg . apache . kafka . connect . source . Source Record Ċ import Ġorg . apache . kafka . connect . source . Source Task Ċ import Ġutil . log Ċ import Ġjava . io . File Ċ import Ġjava . time . Period ĊĊ Ċ data Ġclass ĠConfig ( val Ġpath : String , Ġval Ġtopic : ĠString , Ġval Ġperiod : Period ) Ġ{ ĊĠĠĠ Ġenum Ġclass ĠCONFIG Ġ{ ĠPATH , ĠTOP IC , ĠPER IOD } ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġfun Ġfrom Config ( props : ĠMutable Map < String , ĠString >? ): ĠConfig Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġpath Ġ= Ġprops ?. get ( CONFIG . PATH . toString ()) ?: "" ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtopic Ġ= Ġprops ?. get ( CONFIG . TOPIC . toString ()) ?: "" ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġperiod String Ġ= Ġprops ?. get ( CONFIG . PERIOD . toString ()) ?: "" ĊĠĠĠĠĠĠĠĠĠĠĠ Ġvar Ġperiod : ĠPeriod ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtry { ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġperiod Ġ= ĠPeriod . parse ( period String ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} catch Ġ( e : Exception ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠTODO (" correct Ġlogging ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĠConfig ( path , Ġtopic , Ġperiod ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ Ċ class ĠIn gest Connector : ĠSource Connector () Ġ{ ĊĊĠĠĠ Ġprivate Ġvar Ġconfig : ĠConfig Ġ= ĠConfig ("", Ġ"", ĠPeriod . ZERO ) ĊĊĠĠĠ Ġoverride Ġfun Ġtask Configs ( max Tasks : ĠInt ): ĠMutable List < Mutable Map < String , ĠString >> Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġone Config Ġ= Ġmutable Map Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠConfig . CONFIG . PATH . toString () Ġto Ġconfig . path , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠConfig . CONFIG . TOPIC . toString () Ġto Ġconfig . topic , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠConfig . CONFIG . PERIOD . toString () Ġto Ġconfig . period . toString ()) ĊĠĠĠĠĠĠĠ Ġreturn Ġmutable ListOf ( one Config ) ĊĠĠĠ Ġ} ĊĊĊĊĠĠĠ Ġoverride Ġfun Ġstart ( props : ĠMutable Map < String , ĠString > ?) Ġ{ ĊĠĠĠĠĠĠĠ Ġconfig Ġ= ĠConfig . from Config ( props ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġstop () Ġ{ ĊĠĠĠĠĠĠĠ ĠTODO (" not Ġimplemented ") Ġ// To Ġchange Ġbody Ġof Ġcreated Ġfunctions Ġuse ĠFile Ġ| ĠSettings Ġ| ĠFile ĠTemplates . ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġversion (): ĠString Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġ" 0 . 0 . 1 " ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġtask Class (): ĠClass < out ĠTask > Ġ{ ĊĠĠĠĠĠĠ Ġreturn ĠIn gest Task :: class . java ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġconfig (): ĠConfig Def Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġconfig Ġ= ĠConfig Def () ĊĠĠĠĠĠĠĠ Ġconfig . define ( Config . CONFIG . PATH . toString (), Config Def . Type . STRING , Config Def . Import ance . HIGH ," just Ġa Ġdescription ") ĊĠĠĠĠĠĠĠ Ġreturn Ġconfig ĊĠĠĠ Ġ} Ċ } Ċ Ċ class ĠIn gest Task : ĠSource Task () Ġ{ ĊĠĠĠ Ġprivate Ġvar Ġconfig : ĠConfig Ġ= ĠConfig ("", Ġ"", ĠPeriod . ZERO ) ĊĊĠĠĠ Ġoverride Ġfun Ġstart ( props : ĠMutable Map < String , ĠString > ?) Ġ{ ĊĠĠĠĠĠĠĠ Ġconfig Ġ= ĠConfig . from Config ( props ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġstop () Ġ{ ĊĠĠĠĠĠĠĠ ĠTODO (" not Ġimplemented ") Ġ// To Ġchange Ġbody Ġof Ġcreated Ġfunctions Ġuse ĠFile Ġ| ĠSettings Ġ| ĠFile ĠTemplates . ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġversion (): ĠString Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġ" 0 . 0 . 1 " ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġpoll (): ĠMutable List < Source Record > Ġ{ ĊĠĠĠĠĠĠĠ Ġvar Ġresult Ġ= Ġmutable ListOf < Source Record >() ĊĠĠĠĠĠĠĠ ĠFile ( config . path ). walk Top Down (). forEach Indexed Ġ{ Ġidx , Ġfile Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlog (" ingest Ġ" Ġ+ Ġfile . absolute Path ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsource Partition Ġ= Ġmap Of < String , String >(" filename " Ġto Ġfile . absolute Path ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġsource Offset Ġ= Ġmap Of < String , String >(" position " Ġto Ġ Ġ"" + Ġidx ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresult . add ( Source Record ( source Partition , Ġsource Offset , Ġconfig . topic , ĠSTRING _ SCHEMA , Ġfile . absolute Path )) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġresult ĊĠĠĠ Ġ} Ċ Ċ }