/* Ċ Ġ* ĠLicensed Ġto Ġthe ĠApache ĠSoftware ĠFoundation Ġ( ASF ) Ġunder Ġone Ċ Ġ* Ġor Ġmore Ġcontributor Ġlicense Ġagreements . Ġ ĠSee Ġthe ĠNOTICE Ġfile Ċ Ġ* Ġdistributed Ġwith Ġthis Ġwork Ġfor Ġadditional Ġinformation Ċ Ġ* Ġregarding Ġcopyright Ġownership . Ġ ĠThe ĠASF Ġlicenses Ġthis Ġfile Ċ Ġ* Ġto Ġyou Ġunder Ġthe ĠApache ĠLicense , ĠVersion Ġ 2 . 0 Ġ( the Ċ Ġ* Ġ" License "); Ġyou Ġmay Ġnot Ġuse Ġthis Ġfile Ġexcept Ġin Ġcompliance Ċ Ġ* Ġwith Ġthe ĠLicense . Ġ ĠYou Ġmay Ġobtain Ġa Ġcopy Ġof Ġthe ĠLicense Ġat Ċ Ġ* Ċ Ġ* ĠĠ Ġhttp :// www . apache . org / licenses / LICENSE - 2 . 0 Ċ Ġ* Ċ Ġ* ĠUnless Ġrequired Ġby Ġapplicable Ġlaw Ġor Ġagreed Ġto Ġin Ġwriting , Ċ Ġ* Ġsoftware Ġdistributed Ġunder Ġthe ĠLicense Ġis Ġdistributed Ġon Ġan Ċ Ġ* Ġ" AS ĠIS " ĠBASIS , ĠWITHOUT ĠWARRANTIES ĠOR ĠCONDITIONS ĠOF ĠANY Ċ Ġ* ĠKIND , Ġeither Ġexpress Ġor Ġimplied . Ġ ĠSee Ġthe ĠLicense Ġfor Ġthe Ċ Ġ* Ġspecific Ġlanguage Ġgoverning Ġpermissions Ġand Ġlimitations Ċ Ġ* Ġunder Ġthe ĠLicense . Ċ Ġ*/ Ċ package Ġio . f lood plain . elasticsearch Ċ Ċ import Ġio . f lood plain . kot l ind sl . Abstract Sink Config Ċ import Ġio . f lood plain . kot l ind sl . F lood plain Sink Ċ import Ġio . f lood plain . kot l ind sl . Material ized Config Ċ import Ġio . f lood plain . kot l ind sl . Partial Stream Ċ import Ġio . f lood plain . kot l ind sl . Stream Ċ import Ġio . f lood plain . kot l ind sl . Transformer Ċ import Ġio . f lood plain . reactive . source . topology . Sink Transformer Ċ import Ġio . f lood plain . streams . api . Topic Ċ import Ġio . f lood plain . streams . api . Topology Context Ċ import Ġio . f lood plain . streams . remote join . Topology Constructor Ċ import Ġjava . util . Optional Ċ Ċ fun ĠStream . elastic Search Config ( name : ĠString , Ġuri : ĠString ): ĠElastic Search Sink Config Ġ{ ĊĠĠĠ Ġval Ġc Ġ= ĠElastic Search Sink Config ( topology Context , Ġtopology Constructor , Ġname , Ġuri , Ġthis . topology Context ) ĊĠĠĠ Ġreturn Ġthis . add Sink Configuration ( c ) Ġas ĠElastic Search Sink Config Ċ } Ċ Ċ class ĠElastic Search Sink Config ( override Ġval Ġtopology Context : ĠTop ology Context , Ġoverride Ġval Ġtopology Constructor : ĠTop ology Constructor , Ġval Ġname : ĠString , Ġval Ġuri : ĠString , Ġval Ġcontext : ĠTop ology Context ) Ġ: ĊĠĠĠ ĠAbstract Sink Config () Ġ{ ĊĠĠĠ Ġ// Ġvar Ġsink Task : ĠElasticsearch Sink Task ? Ġ= Ġnull ĊĠĠĠ Ġval Ġmaterial ized Configs : ĠMutable List < Material ized Config > Ġ= Ġmutable ListOf () ĊĠĠĠ Ġprivate Ġvar Ġinstantiated Sink Elements : ĠMap < Topic , ĠMutable List < F lood plain Sink >> ? Ġ= Ġnull ĊĊĠĠĠ Ġoverride Ġfun Ġmaterial ize Connector Config (): ĠList < Material ized Config > Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġmaterial ized Configs ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġsink Task (): ĠAny ? Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġsink Task ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġsink Elements (): ĠMap < Topic , ĠMutable List < F lood plain Sink >> Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġinstantiated Sink Elements Ġ?: Ġempty Map () ĊĠĠĠ Ġ} Ċ } Ċ Ċ fun ĠPartial Stream . elastic Search Sink ( sink Name : ĠString , Ġtopic Name : ĠString , Ġconfig : ĠElastic Search Sink Config ) Ġ{ ĊĠĠĠ Ġval Ġtopic Ġ= ĠTopic . from ( topic Name , Ġtopology Context ) ĊĠĠĠ Ġval Ġsink Transformer Ġ= ĠS ink Transformer ( Optional . of ( sink Name ), Ġtopic , ĠOptional . empty (), ĠTopic . F lood plain Key Format . FLO OD PLAIN _ STRING , ĠTopic . F lood plain Body Format . CONNECT _ JSON ) ĊĠĠĠ Ġadd Transformer ( Transformer ( root Topology , Ġsink Transformer , Ġtopology Context )) ĊĊĠĠĠ Ġval Ġsink Config Ġ= Ġmap Of ( ĊĠĠĠĠĠĠĠ Ġ" connector . class " Ġto Ġ" io . con fluent . connect . elasticsearch . Elasticsearch Sink Connector ", ĊĠĠĠĠĠĠĠ Ġ" connection . url " Ġto Ġconfig . uri , ĊĠĠĠĠĠĠĠ Ġ" tasks . max " Ġto Ġ" 1 ", ĊĠĠĠĠĠĠĠ Ġ" type . name " Ġto Ġ"_ doc ", ĊĠĠĠĠĠĠĠ Ġ" value . converter " Ġto Ġ" org . apache . kafka . connect . json . Json Converter ", ĊĠĠĠĠĠĠĠ Ġ// Ġ" key . converter " Ġto Ġ" org . apache . kafka . connect . json . Json Converter ", Ġ// Ġmaps Ġnot Ġsupported Ġby Ġelasticsearch ĊĠĠĠĠĠĠĠ Ġ" key . converter " Ġto Ġ" org . apache . kafka . connect . storage . String Converter ", Ġ// Ġmaps Ġnot Ġsupported Ġby Ġelasticsearch ĊĠĠĠĠĠĠĠ Ġ" topics " Ġto Ġtopic . qualified String (), ĊĠĠĠĠĠĠĠ Ġ" schema . ignore " Ġto Ġ" true ", ĊĠĠĠĠĠĠĠ Ġ" value . converter . schemas . enable " Ġto Ġ" false ", ĊĠĠĠĠĠĠĠ Ġ" key . converter . schemas . enable " Ġto Ġ" false ", ĊĠĠĠĠĠĠĠ Ġ" behavior . on . null . values " Ġto Ġ" delete ", ĊĠĠĠĠĠĠĠ Ġ" type . name " Ġto Ġ"_ doc " ĊĠĠĠ Ġ) ĊĠĠĠ Ġconfig . material ized Configs . add ( Material ized Config ( config . name , ĠlistOf ( topic ), Ġsink Config )) ĊĠĠĠ Ġ// Ġval Ġconn Ġ= ĠElasticsearch Sink Connector () ĊĠĠĠ Ġ// Ġconn . start ( sink Config ) ĊĠĠĠ Ġ// Ġval Ġtask Ġ= Ġconn . task Class (). getDeclared Constructor (). newInstance () Ġas ĠElasticsearch Sink Task ĊĠĠĠ Ġ// Ġtask . start ( sink Config ) ĊĠĠĠ Ġ// Ġconfig . sink Task Ġ= Ġtask ĊĠĠĠ Ġ// Ġreturn Ġf lood plain Sink From Task ( task , Ġconfig ) Ċ } Ċ