package Ġno . nav . hel se Ċ Ċ import Ġcom . github . kit tin un f . fuel . http Get Ċ import Ġio . prometheus . client . Collector Registry Ċ import Ġno . nav . common . JA AS Credential Ċ import Ġno . nav . common . Kafka Environment Ċ import Ġno . nav . hel se . streams . Json Deserializer Ċ import Ġno . nav . hel se . streams . Json Serializer Ċ import Ġno . nav . hel se . streams . Topics Ċ import Ġorg . apache . kafka . clients . Common Client Configs Ċ import Ġorg . apache . kafka . clients . consumer . Consumer Config Ċ import Ġorg . apache . kafka . clients . consumer . Kafka Consumer Ċ import Ġorg . apache . kafka . clients . producer . Kafka Producer Ċ import Ġorg . apache . kafka . clients . producer . Producer Config Ċ import Ġorg . apache . kafka . clients . producer . Producer Record Ċ import Ġorg . apache . kafka . common . config . S asl Configs Ċ import Ġorg . json . JSONObject Ċ import Ġorg . junit . jupiter . api . After All Ċ import Ġorg . junit . jupiter . api . After Each Ċ import Ġorg . junit . jupiter . api . Assertions Ċ import Ġorg . junit . jupiter . api . Assertions . assertEquals Ċ import Ġorg . junit . jupiter . api . Before All Ċ import Ġorg . junit . jupiter . api . BeforeEach Ċ import Ġorg . junit . jupiter . api . Test Ċ import Ġjava . time . Duration Ċ import Ġjava . util . Properties Ċ Ċ class ĠComponent Test Ġ{ ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval Ġusername Ġ= Ġ" srv k af k ac lient " ĊĠĠĠĠĠĠĠ Ġprivate Ġconst Ġval Ġpassword Ġ= Ġ" k af k ac lient " ĊĊĠĠĠĠĠĠĠ Ġval Ġembedded Environment Ġ= ĠKafka Environment ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġusers Ġ= ĠlistOf ( JA AS Credential ( username , Ġpassword )), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġauto Start Ġ= Ġfalse , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwith Schema Registry Ġ= Ġfalse , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwith Security Ġ= Ġtrue , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtopics Ġ= ĠlistOf ( Topics . SY KE PEN GE BE HAND LING . name , ĠTop ics . SY KE PEN GES Ã ĺ KN ADER _ IN N . name ) ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġenv Ġ= ĠEnvironment ( username Ġ= Ġusername , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpassword Ġ= Ġpassword , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġbootstrap Servers Url Ġ= Ġembedded Environment . bro kers URL ) ĊĊĠĠĠĠĠĠĠ Ġ@ Before All ĊĠĠĠĠĠĠĠ Ġ@ Jvm Static ĊĠĠĠĠĠĠĠ Ġfun Ġsetup () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠCollector Registry . default Registry . clear () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġembedded Environment . start () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġ@ After All ĊĠĠĠĠĠĠĠ Ġ@ Jvm Static ĊĠĠĠĠĠĠĠ Ġfun Ġteardown () Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġembedded Environment . te ar Down () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġspin ne : ĠSpin ne ĊĊĠĠĠ Ġ@ BeforeEach ĊĠĠĠ Ġfun Ġ` start Ġspin ne `() Ġ{ ĊĠĠĠĠĠĠĠ Ġspin ne Ġ= ĠSpin ne ( env ) ĊĠĠĠĠĠĠĠ Ġspin ne . start () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ After Each ĊĠĠĠ Ġfun Ġ` clear Ġprometheus Ġregistry `() Ġ{ ĊĠĠĠĠĠĠĠ Ġspin ne . stop () ĊĠĠĠĠĠĠĠ ĠCollector Registry . default Registry . clear () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` embedded Ġkafka Ġcluster Ġis Ġup Ġand Ġrunning `() Ġ{ ĊĠĠĠĠĠĠĠ ĠassertEquals ( embedded Environment . server Par k . status , ĠKafka Environment . Server Par k Status . Started ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` message Ġfrom Ġsy fo Ġis Ġnot Ġforwarded Ġbecause Ġstatus Ġis Ġnot Ġsend t `() Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmsg Sent Ġ= ĠJSONObject (""" {" ak tor Id ": Ġ" 1 5 7 3 0 8 2 1 8 6 6 9 9 ", Ġ" so kn ad stype ": Ġ" ty pen ", Ġ" status ": Ġ" ny "} """) ĊĠĠĠĠĠĠĠ Ġproduce One Message ( msg Sent ) ĊĊĠĠĠĠĠĠĠ Ġval Ġresult Consumer Ġ= ĠKafka Consumer < String , ĠJSONObject >( consumer Properties ()) ĊĠĠĠĠĠĠĠ Ġresult Consumer . subscribe ( listOf ( Topics . SY KE PEN GE BE HAND LING . name )) ĊĠĠĠĠĠĠĠ Ġval Ġconsumer Records Ġ= Ġresult Consumer . poll ( Duration . of Seconds ( 1 0 )) ĊĊĠĠĠĠĠĠĠ ĠassertEquals ( 0 , Ġconsumer Records . count ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` message Ġfrom Ġsy fo Ġis Ġforwarded Ġas Ġis Ġto Ġbe handling Ġtopic `() Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmsg Sent Ġ= ĠJSONObject (""" {" ak tor Id ": Ġ" 1 5 7 3 0 8 2 1 8 6 6 9 9 ", Ġ" so kn ad stype ": Ġ" ty pen ", Ġ" status ": Ġ" send t "} """) ĊĠĠĠĠĠĠĠ Ġproduce One Message ( msg Sent ) ĊĊĠĠĠĠĠĠĠ Ġval Ġresult Consumer Ġ= ĠKafka Consumer < String , ĠJSONObject >( consumer Properties ()) ĊĠĠĠĠĠĠĠ Ġresult Consumer . subscribe ( listOf ( Topics . SY KE PEN GE BE HAND LING . name )) ĊĠĠĠĠĠĠĠ Ġval Ġconsumer Records Ġ= Ġresult Consumer . poll ( Duration . of Seconds ( 1 0 )) ĊĊĠĠĠĠĠĠĠ ĠassertEquals ( 1 , Ġconsumer Records . count ()) ĊĊĠĠĠĠĠĠĠ Ġval Ġmsg Read Ġ= Ġconsumer Records . records ( Topics . SY KE PEN GE BE HAND LING . name ). first (). value () ĊĠĠĠĠĠĠĠ Ġmsg Read . keySet (). forEach Ġ{ Ġprop Name Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ ĠassertEquals ( msg Sent [ prop Name ], Ġmsg Read [ prop Name ]) ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġverify Metrics () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġproducer Properties (): ĠProperties Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠProperties (). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Consumer Config . BOOT STR AP _ SERVER S _ CONFIG , Ġembedded Environment . bro kers URL ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Producer Config . KEY _ SERIAL IZER _ CLASS _ CONFIG , Ġ" org . apache . kafka . common . serialization . String Serializer ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Producer Config . VALUE _ SERIAL IZER _ CLASS _ CONFIG , ĠJsonSerializer :: class . java ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Common Client Configs . SECURITY _ PROTOCOL _ CONFIG , Ġ" SAS L _ PL AINT EXT ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( S asl Configs . SAS L _ ME CHAN IS M , Ġ" PLAIN ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( S asl Configs . SAS L _ JA AS _ CONFIG , Ġ" org . apache . kafka . common . security . plain . Plain Login Module Ġrequired Ġusername =\" $ username \" Ġpassword =\" $ password \"; ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġconsumer Properties (): ĠProperties Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn ĠProperties (). apply Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Consumer Config . BOOT STR AP _ SERVER S _ CONFIG , Ġembedded Environment . bro kers URL ) ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Consumer Config . KEY _ DE SERIAL IZER _ CLASS _ CONFIG , Ġ" org . apache . kafka . common . serialization . String Deserializer ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Consumer Config . VALUE _ DE SERIAL IZER _ CLASS _ CONFIG , ĠJson Deserializer :: class . java ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Common Client Configs . SECURITY _ PROTOCOL _ CONFIG , Ġ" SAS L _ PL AINT EXT ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( S asl Configs . SAS L _ ME CHAN IS M , Ġ" PLAIN ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( S asl Configs . SAS L _ JA AS _ CONFIG , Ġ" org . apache . kafka . common . security . plain . Plain Login Module Ġrequired Ġusername =\" $ username \" Ġpassword =\" $ password \"; ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġput ( Consumer Config . GROUP _ ID _ CONFIG , Ġ" spin ne - test - verification ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġproduce One Message ( message : ĠJSONObject ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġproducer Ġ= ĠKafka Producer < String , ĠJSONObject >( producer Properties ()) ĊĠĠĠĠĠĠĠ Ġproducer . send ( Producer Record ( Topics . SY KE PEN GES Ã ĺ KN ADER _ IN N . name , Ġnull , Ġmessage )) ĊĠĠĠĠĠĠĠ Ġproducer . flush () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġverify Metrics () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġ(_, Ġresponse , Ġ_) Ġ= Ġ" http :// localhost : 8 0 8 0 / metrics ". http Get (). response () ĊĠĠĠĠĠĠĠ Ġval Ġakt or Regex Ġ= Ġ" .* status =\" (.* ) \" .* Ġ( \\ d * \\. \\ d *) $ ". to Regex () ĊĠĠĠĠĠĠĠ Ġval Ġcounters Ġ= ĠString ( response . data ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. lines () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġit . startsWith (" sy ke p enger _ mot tat te _ so kn ader { type =") Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġakt or Regex . match Ent ire ( it ) !! Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ ĠPair ( first Ġ= Ġit . group Values [ 1 ], Ġsecond Ġ= Ġit . group Values [ 2 ]) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. toMap () ĊĠĠĠĠĠĠĠ ĠAssertions . assertEquals (" 1 . 0 ", Ġcounters [" send t "]) ĊĠĠĠ Ġ} Ċ } ĊĊ