package Ġio . f per ez p . connect . mock Ċ Ċ import Ġarrow . fx . coroutines . Atomic Ċ import Ġio . ktor . application .* Ċ import Ġio . ktor . features .* Ċ import Ġio . ktor . http .* Ċ import Ġio . ktor . jackson .* Ċ import Ġio . ktor . request .* Ċ import Ġio . ktor . response .* Ċ import Ġio . ktor . routing .* Ċ import Ġio . ktor . server . engine .* Ċ import Ġio . ktor . server . netty .* Ċ import Ġio . ktor . util .* Ċ import Ġjava . util .* ĊĊ Ċ data Ġclass ĠConnector CreateRequest ( val Ġname : ĠString , Ġval Ġconfig : ĠMap < String , ĠString >) Ċ Ċ data Ġclass ĠConnector Response ( ĊĠĠĠ Ġval Ġname : ĠString , ĊĠĠĠ Ġval Ġconfig : ĠMap < String , ĠString >, ĊĠĠĠ Ġval Ġtasks : ĠList < Connector Task >, Ċ ) Ċ Ċ data Ġclass ĠConnector Status Response ( val Ġname : ĠString , Ġval Ġtasks : ĠList < Task Status >) Ċ Ċ // ĠState Ġto Ġbe Ġmoved Ġto Ġan ĠEnum Ġwhen Ġworking Ġon Ġhttps :// l ando op . atlassian . net / browse / FLO WS - 1 3 6 5 Ċ // ĠThis Ġhas Ġbeen Ġcreated Ġnow Ġfor Ġtesting ĠDeploy ments , Ġbut Ġit 's Ġnot Ġa Ġfinal Ġversion Ġand Ġwill Ġbe Ġre work ed Ċ // Ġas Ġpart Ġof Ġthe Ġticket Ġmentioned Ġabove . Ċ data Ġclass ĠTask Status ( ĊĠĠĠ Ġval Ġid : ĠInt , ĊĠĠĠ Ġval Ġstate : ĠConnect Status , ĊĠĠĠ Ġval Ġworker _ id : ĠString , ĊĠĠĠ Ġval Ġtrace : ĠString ? Ġ= Ġnull , Ċ ) Ċ Ċ data Ġclass ĠInfo ( val Ġversion : ĠString , Ġval Ġcommit : ĠString , Ġval Ġkafka _ cluster _ id : ĠString ) Ċ Ċ data Ġclass ĠPlugin ( val Ġ` class `: ĠString , Ġval Ġ` type `: ĠString , Ġval Ġversion : ĠString ) Ċ Ċ data Ġclass ĠInvalid Connect Response ( val Ġmessage : ĠString , Ġval Ġcode : ĠInt , Ġval Ġerror : ĠThrowable ? Ġ= Ġnull ) Ċ Ċ data Ġclass ĠConnector Task ( val Ġconnector : ĠString , Ġval Ġtask : ĠInt ) Ċ data Ġclass ĠConnector Error ( val Ġcluster Name : ĠString , Ġval ĠstatusCode : ĠInt , Ġval Ġmessage : ĠError Message ) Ċ data Ġclass ĠError Message ( val Ġmessage : ĠString ) Ċ data Ġclass ĠConnect Info ( val Ġname : ĠString , Ġval Ġversion : ĠString ) Ċ data Ġclass ĠConnector Plugin ( val ĠclassName : ĠString , Ġval Ġplugin Type : ĠString , Ġval Ġversion : ĠString ) ĊĊ Ċ enum Ġclass ĠConnect Status Ġ{ ĊĠĠĠ ĠRUN NING , ĠPA USED , ĠFAILED , ĠUN ASS IGNED Ċ } Ċ Ċ data Ġclass ĠConfig ( val Ġworker AP ort : ĠInt , Ġval Ġworker B Port : ĠInt , Ġval Ġworker CP ort : ĠInt ) Ċ Ċ fun Ġmain () Ġ{ ĊĠĠĠ Ġval Ġconfig Ġ= ĠConfig ( 8 0 8 3 , Ġ 8 1 8 3 , Ġ 8 2 8 3 ) Ċ Ċ // ĠĠĠ Ġval Ġconfig Ġ= ĠConfig Loader (). load Config < Config >(" application . conf "). get Unsafe () ĊĊĠĠĠ Ġval Ġenv Ġ= Ġapplication Engine Environment Ġ{ ĊĠĠĠĠĠĠĠ Ġmodule Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmain () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ// ĠWorker A ĊĠĠĠĠĠĠĠ Ġconnector Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġhost Ġ= Ġ" 0 . 0 . 0 . 0 " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġport Ġ= Ġconfig . worker AP ort ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ// ĠWorker B ĊĠĠĠĠĠĠĠ Ġconnector Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġhost Ġ= Ġ" 0 . 0 . 0 . 0 " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġport Ġ= Ġconfig . worker B Port ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ// ĠWorker C ĊĠĠĠĠĠĠĠ Ġconnector Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġhost Ġ= Ġ" 0 . 0 . 0 . 0 " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġport Ġ= Ġconfig . worker CP ort ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĠĠĠ Ġembedded Server ( Net ty , Ġenv ). start ( true ) Ċ } Ċ Ċ fun ĠApplication . main () Ġ{ ĊĊĠĠĠ Ġval Ġcluster Id : ĠString Ġ= ĠUUID . randomUUID (). toString () ĊĠĠĠ Ġval Ġconnector s Ġ= ĠAtomic . unsafe ( empty Map < String , ĠMap < String , ĠString >> ()) ĊĊĠĠĠ Ġconfigure Json Serialization () ĊĠĠĠ Ġconfigure Error Interceptor () ĊĠĠĠ Ġinstall ( Call Logging ) ĊĠĠĠ Ġinstall ( Default Headers ) ĊĠĠĠ Ġinstall ( Auto Head Response ) ĊĠĠĠ Ġinstall ( X Forwarded Header Support ) ĊĠĠĠ Ġinstall ( Forwarded Header Support ) ĊĊĠĠĠ Ġrouting Ġ{ ĊĠĠĠĠĠĠĠ Ġtrace Ġ{ Ġapplication . log . trace ( it . build Text ()) Ġ} ĊĊĠĠĠĠĠĠĠ Ġroute ("/") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġget Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcall . respond ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠHttpStatusCode . OK , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠInfo (" 2 . 5 . 1 - L 0 ", Ġ" 0 efa 8 fb 0 f 4 c 7 3 d 9 2 ", Ġ" Sr 6 u NB j PR N - AH 8 B 1 0 Et F 2 A ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĊĠĠĠĠĠĠĠ Ġroute ("/ connector - plugins ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġget Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcall . respond ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠHttpStatusCode . OK , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠlistOf ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠPlugin (" io . l enses . runner . connect . Processor Connector ", Ġ" sink ", Ġ" mock "), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠPlugin (" com . l ando op . connect . SQL ", Ġ" sink ", Ġ" mock ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġroute ("/ connectors ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpost Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġrequest Ġ= Ġcall . receive < Connector CreateRequest >() ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconnector s . update Ġ{ Ġ( it Ġ+ Ġ( request . name Ġto Ġrequest . config )) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtasks Max : ĠInt Ġ= Ġrequest . config [" tasks . max "] ?. toInt () Ġ?: Ġ 0 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcall . respond ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠHttpStatusCode . Created , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠConnector Response ( request . name , Ġrequest . config , Ġbuild Tasks List ( request . name , Ġtasks Max )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġget Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġconnector Names Ġ= Ġconnector s . get (). keys ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcall . respond ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠHttpStatusCode . OK , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconnector Names ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġroute ("{ connector Name }") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġget (" status ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġconnector Name Ġ= Ġcall . parameters . get OrFail (" connector Name ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtasks Max : ĠInt Ġ= Ġconnector s . get ()[ connector Name ]?. get (" tasks . max ") ?. toInt () Ġ?: Ġ 0 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcall . respond ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠHttpStatusCode . OK , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠConnector Status Response ( connector Name , Ġbuild Tasks Details List ( tasks Max )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġroute (" config ") Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġput Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġrequest Ġ= Ġcall . receive < Connector CreateRequest >() ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconnector s . update Ġ{ Ġ( it Ġ+ Ġ( request . name Ġto Ġrequest . config )) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġtasks Max : ĠInt Ġ= Ġrequest . config [" tasks . max "] ?. toInt () Ġ?: Ġ 0 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġconnector Name Ġ= Ġcall . parameters . get OrFail (" connector Name ") ĊĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcall . respond ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠHttpStatusCode . OK , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠConnector Response ( request . name , Ġrequest . config , Ġbuild Tasks List ( connector Name , Ġtasks Max )) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġget Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġconnector Name Ġ= Ġcall . parameters . get OrFail (" connector Name ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġconnector Config : ĠMap < String , ĠString > Ġ= Ġconnector s . get ()[ connector Name ] Ġ?: Ġempty Map () ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcall . respond ( HttpStatusCode . OK , Ġconnector Config ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ Ċ private Ġfun Ġbuild Tasks Details List ( tasks Max : ĠInt ) Ġ= ĠList ( tasks Max ) Ġ{ ĊĠĠĠ ĠTask Status ( it , ĠConnect Status . RUNNING , Ġ" 1 2 7 . 0 . 0 . 1 ") Ċ } Ċ Ċ private Ġfun Ġbuild Tasks List ( connector Name : ĠString , Ġtasks Max : ĠInt ) Ġ= ĠList ( tasks Max ) Ġ{ ĊĠĠĠ ĠConnector Task ( connector Name , Ġit ) Ċ } Ċ Ċ private Ġfun ĠApplication . configure Error Interceptor () Ġ{ ĊĠĠĠ Ġinstall ( Status Pages ) Ġ{ ĊĠĠĠĠĠĠĠ Ġexception < NotFoundException > Ġ{ Ġcause Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcall . respond ( HttpStatusCode . NotFound , Ġ" Not Ġfound : Ġ${ cause . message }") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġcause ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġexception < Missing Request Parameter Exception > Ġ{ Ġcause Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcall . respond ( HttpStatusCode . NotFound , Ġ" Not Ġfound : Ġ${ cause . message }") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġcause ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġexception < Throwable > Ġ{ Ġcause Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcall . respond ( HttpStatusCode . InternalServerError , Ġ" Internal ĠServer ĠError : Ġ${ cause . message }") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow Ġcause ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ Ċ private Ġfun ĠApplication . configure Json Serialization () Ġ{ ĊĠĠĠ Ġinstall ( Content Neg otiation ) Ġ{ ĊĠĠĠĠĠĠĠ Ġregister ( ContentType . Application . Json , ĠJack son Converter ()) ĊĠĠĠ Ġ} Ċ }