package Ġcom . program mable production . kafka . kotlin . testing Ċ Ċ import Ġarrow .* Ċ import Ġarrow . core .* Ċ import Ġjava . net . Socket Ċ import Ġjava . net . Socket Exception Ċ import Ġjava . util . Properties Ċ Ċ import Ġcom . types afe . scal alog ging . Strict Logging Ċ import Ġio . con fluent . kafka . sche m are gistry . avro . Av ro Compatibility Level Ċ import Ġio . con fluent . kafka . sche m are gistry . client . rest . Rest Service Ċ import Ġio . con fluent . kafka . sche m are gistry . rest . Schema Registry Config Ċ import Ġio . con fluent . kafka . sche m are gistry . rest . Schema Registry Rest Application Ċ import Ġio . con fluent . kafka . sche m are gistry . storage . Schema Registry Ċ import Ġio . con fluent . kafka . sche m are gistry . storage . Schema Registry Identity Ċ import Ġorg . eclipse . jetty . server . Server Ċ Ċ class ĠSchema Registry Service ( val Ġport : ĠInt , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġz ookeeper Connection : ĠString , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġkafka Topic : ĠString , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġav ro Compatibility Level : ĠAv ro Compatibility Level , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġmaster El ig ibility : ĠBoolean ){ ĊĊĠĠĠ Ġval Ġschema Registry Prop Ġ= Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġprop Ġ= ĠProperties () ĊĠĠĠĠĠĠĠ Ġprop . setProperty (" port ", Ġport . toString ()) ĊĠĠĠĠĠĠĠ Ġprop . setProperty ( Schema Registry Config . K AF K AST ORE _ CONNECTION _ URL _ CONFIG , Ġz ookeeper Connection ) ĊĠĠĠĠĠĠĠ Ġprop . put ( Schema Registry Config . K AF K AST ORE _ TOPIC _ CONFIG , Ġkafka Topic ) ĊĠĠĠĠĠĠĠ Ġprop . put ( Schema Registry Config . COMPAT IBILITY _ CONFIG , Ġav ro Compatibility Level . toString ()) ĊĠĠĠĠĠĠĠ Ġprop . put ( Schema Registry Config . MASTER _ EL IG IBILITY , Ġmaster El ig ibility ) ĊĠĠĠĠĠĠĠ Ġprop ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġval Ġapp Ġ= ĠSchema Registry Rest Application ( schema Registry Prop ()) ĊĊĠĠĠ Ġval Ġrest Server Ġ= Ġstart Server ( port ) ĊĊĠĠĠ Ġvar ĠEndpoint : ĠString Ġ= Ġget Endpoint ( rest Server ) ĊĊĠĠĠ Ġval Ġrest Client : ĠRest Service Ġ= ĠRest Service ( Endpoint ) ĊĊĠĠĠ Ġfun Ġstart Server ( port : ĠInt , Ġretries : ĠInt Ġ= Ġ 5 ): ĠOption < Server > Ġ{ ĊĠĠĠĠĠĠĠ Ġvar Ġretry Ġ= Ġretries Ġ> Ġ 0 ĊĠĠĠĠĠĠĠ Ġvar Ġrest Server : ĠOption < Server > Ġ= ĠNone ĊĊĠĠĠĠĠĠĠ Ġif Ġ( retry ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( is Port InUse ( port )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprintln (" Schema ĠRegistry ĠPort Ġ$ port Ġis Ġalready Ġin Ġuse ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠThread . sleep ( 2 0 0 0 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġstart Server ( port , Ġretries Ġ- Ġ 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġserver : ĠServer Ġ= Ġapp . create Server () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġserver . start () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrest Server Ġ= ĠSome ( server ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġrest Server ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġget Endpoint ( rest Server : ĠOption < Server >) = ĊĠĠĠĠĠĠĠ Ġwhen ( rest Server ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠSome Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġuri Ġ= Ġrest Server . t . uri . toString () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( uri . endsWith ("/ ")) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġuri . substring ( 0 , Ġuri . length Ġ- Ġ 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġuri ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠNone Ġ-> Ġ"" ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġis Port InUse ( port : ĠInt )= Ġtry Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠSocket (" 1 2 7 . 0 . 0 . 1 ", Ġport ). close () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcatch Ġ( e : ĠException ){ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen ( e ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠSocket Exception Ġ-> Ġfalse ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĊĠĠĠ Ġfun Ġclose () Ġ= ĊĠĠĠĠĠĠĠ Ġwhen Ġ( rest Server ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠSome Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrest Server . t . stop () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrest Server . t . join () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġis ĠNone Ġ-> ĠUnit ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġval Ġis Master : ĠBoolean Ġ= Ġapp . schema Registry (). is Master ĊĊĠĠĠ Ġfun Ġset Master ( schema Registry Identity : ĠSchema Registry Identity ) Ġ= Ġapp . schema Registry (). set Master ( schema Registry Identity ) ĊĊĠĠĠ Ġval Ġmy Identity : ĠSchema Registry Identity Ġ= Ġapp . schema Registry (). my Identity () ĊĊĠĠĠ Ġval Ġmaster Identity : ĠSchema Registry Identity Ġ= Ġapp . schema Registry (). master Identity () ĊĊĠĠĠ Ġval Ġschema Registry : ĠSchema Registry Ġ= Ġapp . schema Registry () Ċ } Ċ