/* Ċ Ġ* Ġ============ LICENSE _ START ================================================ ======= Ċ Ġ* Ġd cae gen 2 - collect ors - ves hv Ċ Ġ* Ġ================================================================= =============== Ċ Ġ* ĠCopyright Ġ( C ) Ġ 2 0 1 9 ĠN OK IA Ċ Ġ* Ġ================================================================= =============== Ċ Ġ* ĠLicensed Ġ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 . Ċ Ġ* Ġ============ LICENSE _ END ================================================ ========= Ċ Ġ*/ Ċ package Ġorg . on ap . dca e . collect ors . ves hv . impl . adapters . kafka Ċ Ċ import Ġorg . apache . kafka . clients . Common Client Configs Ċ import Ġorg . apache . kafka . clients . producer . Producer Config Ċ import Ġorg . apache . kafka . common . config . S asl Configs Ċ import Ġorg . apache . kafka . common . security . auth . Security Protocol Ċ import Ġorg . apache . kafka . common . security . plain . internals . Plain S asl Server Ċ import Ġorg . jetbrains . annotations . Nullable Ċ import Ġorg . on ap . dca e . collect ors . ves hv . domain . V es Message Ċ import Ġorg . on ap . dca e . collect ors . ves hv . utils . apply If Ċ import Ġorg . on ap . dca e gen 2 . services . sdk . model . streams . A af Credentials Ċ import Ġorg . on ap . dca e gen 2 . services . sdk . model . streams . dma ap . Kafka Sink Ċ import Ġorg . on ap . ves . V es Event OuterClass . Common Event Header Ċ import Ġreactor . kafka . sender . Sender Options Ċ Ċ internal Ġobject ĠKafka Sender Options Factory Ġ{ ĊĊĠĠĠ Ġprivate Ġconst Ġval ĠMAX IMUM _ REQUEST _ SIZE _ MULTI PLIER Ġ= Ġ 1 . 2 f ĊĠĠĠ Ġprivate Ġconst Ġval ĠBUFFER _ MEMORY _ MULTI PLIER Ġ= Ġ 3 2 ĊĠĠĠ Ġprivate Ġconst Ġval ĠMIN IMUM _ BUFFER _ MEMORY Ġ= Ġ 3 2 Ġ* Ġ 1 0 2 4 Ġ* Ġ 1 0 2 4 ĊĊĠĠĠ Ġprivate Ġconst Ġval ĠLOGIN _ MODULE _ CLASS Ġ= Ġ" org . apache . kafka . common . security . plain . Plain Login Module " ĊĠĠĠ Ġprivate Ġval ĠSAS L _ PL AINT EXT Ġ= Ġ( Security Protocol . SAS L _ PL AINT EXT Ġas ĠEnum < Security Protocol >). name ĊĊĠĠĠ Ġfun Ġcreate Sender Options ( kafka Sink : ĠKafka Sink ): ĠSender Options < Common Event Header , ĠV es Message > Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠ ĠSender Options . create < Common Event Header , ĠV es Message >() ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. producer Property ( Producer Config . BOOT STR AP _ SERVER S _ CONFIG , Ġkafka Sink . bootstrap Servers ()) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. producer Property ( Producer Config . MAX _ REQUEST _ SIZE _ CONFIG , Ġmax Request Size ( kafka Sink )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. producer Property ( Producer Config . BUFFER _ MEMORY _ CONFIG , Ġbuffer Memory ( kafka Sink )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. producer Property ( Producer Config . KEY _ SERIAL IZER _ CLASS _ CONFIG , ĠProtobuf Serializer :: class . java ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. producer Property ( Producer Config . VALUE _ SERIAL IZER _ CLASS _ CONFIG , ĠV es Message Serializer :: class . java ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. producer Property ( Producer Config . MAX _ IN _ FL IGHT _ REQUEST S _ PER _ CONNECTION , Ġ 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. producer Property ( Producer Config . RE TR IES _ CONFIG , Ġ 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. producer Property ( Producer Config . ACK S _ CONFIG , Ġ" 1 ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. stop OnError ( false ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. apply If ( kafka Sink . aaf Credentials () Ġ!= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġproducer Property ( Common Client Configs . SECURITY _ PROTOCOL _ CONFIG , ĠSAS L _ PL AINT EXT ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. producer Property ( S asl Configs . SAS L _ ME CHAN IS M , ĠPlain S asl Server . PLAIN _ ME CHAN IS M ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. producer Property ( S asl Configs . SAS L _ JA AS _ CONFIG , Ġja as Config ( kafka Sink . aaf Credentials () !! )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġja as Config ( aaf Credentials : ĠA af Credentials ) Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ""" $ LOGIN _ MODULE _ CLASS Ġrequired Ġusername ="${ aaf Credentials . username (). ja as Escape () }" Ġpassword ="${ aaf Credentials . password (). ja as Escape ()} "; """ ĊĊĠĠĠ Ġprivate Ġfun ĠString ?. ja as Escape () Ġ= Ġthis ?. replace (" \"", Ġ"\\ \" ") ĊĊĠĠĠ Ġprivate Ġfun Ġmax Request Size ( kafka Sink : ĠKafka Sink ) Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ( MAXIMUM _ REQUEST _ SIZE _ MULTI PLIER Ġ* Ġkafka Sink . max Payload Size Bytes ()). toInt () ĊĊĠĠĠ Ġprivate Ġfun Ġbuffer Memory ( kafka Sink : ĠKafka Sink ) Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠ ĠInteger . max ( MIN IMUM _ BUFFER _ MEMORY , ĠBUFFER _ MEMORY _ MULTI PLIER Ġ* Ġkafka Sink . max Payload Size Bytes ()) Ċ Ċ } Ċ