package Ġno . nav . hel se . slow tests . kafka Ċ Ċ import Ġkotlinx . coroutines . run Blocking Ċ import Ġno . nav . hel se . sp ion . ved t aks meld ing . Sp le is Mel ding stype Ċ import Ġno . nav . hel se . sp ion . ved t aks meld ing . Sp le is V ed t aks meld ing Generator Ċ import Ġno . nav . hel se . sp ion . ved t aks meld ing . V ed t aks meld ing Client Ċ import Ġno . nav . hel se . sp ion . web . common Ċ import Ġorg . apache . kafka . clients . admin . Admin Client Ċ import Ġorg . apache . kafka . clients . admin . Kafka Admin Client Ċ import Ġorg . apache . kafka . clients . admin . New Topic Ċ import Ġorg . apache . kafka . clients . producer . Kafka Producer Ċ import Ġorg . apache . kafka . clients . producer . Producer Record Ċ import Ġorg . apache . kafka . common . header . internals . Record Header Ċ import Ġorg . apache . kafka . common . serialization . String Serializer Ċ import Ġorg . assertj . core . api . Assertions . assertThat Ċ import Ġorg . assertj . core . api . Assertions . assertThat Exception OfType Ċ import Ġorg . junit . jupiter . api . After All Ċ import Ġorg . junit . jupiter . api . Before All Ċ import Ġorg . junit . jupiter . api . Test Ċ import Ġorg . junit . jupiter . api . Test Instance Ċ import Ġorg . k oin . core . K oin Application Ċ import Ġorg . k oin . core . component . K oin Component Ċ import Ġjava . util . concurrent . TimeUnit Ċ Ċ /** Ċ Ġ* ĠDis se Ġtest ene Ġk re ver Ġen Ġk j Ã¸ ren de ĠKafka Ġbroker ĠpÃ¥ Ġlocalhost : 9 0 9 2 Ċ Ġ* ĠFor ĠÃ¥ Ġk j Ã¸ re Ġopp Ġen Ġkan Ġdu Ġg j Ã¸ re Ċ Ġ* Ġcd Ġdocker / local Ċ Ġ* Ġdocker - compose Ġbuild Ċ Ġ* Ġdocker - compose Ġup Ċ Ġ*/ Ċ @ Test Instance ( Test Instance . Lifecycle . PER _ CLASS ) Ċ internal Ġclass ĠV ed t aks meld ing Client Test Ġ: ĠK oin Component Ġ{ ĊĠĠĠ Ġprivate Ġlateinit Ġvar Ġadmin Client : ĠAdmin Client ĊĠĠĠ Ġval Ġtopic Name Ġ= Ġ" topic " ĊĠĠĠ Ġlateinit Ġvar Ġk oin : ĠK oin Application ĊĠĠĠ Ġval Ġgenerator Ġ= ĠSp le is V ed t aks meld ing Generator ( max Unique Ar be id sg iv ere Ġ= Ġ 1 0 , Ġmax Unique Person er Ġ= Ġ 1 0 ) ĊĊĠĠĠ Ġval Ġtest Props Ġ= Ġmutable Map Of < String , ĠAny >( ĊĠĠĠĠĠĠĠ Ġ" bootstrap . servers " Ġto Ġ" localhost : 9 0 9 2 ", ĊĠĠĠĠĠĠĠ Ġ" max . poll . interval . ms " Ġto Ġ" 3 0 0 0 0 " ĊĠĠĠ Ġ) ĊĊĠĠĠ Ġ@ Before All ĊĠĠĠ Ġinternal Ġfun ĠsetUp () Ġ{ ĊĠĠĠĠĠĠĠ Ġk oin Ġ= ĠK oin Application . Companion . init (). modules ( common ) ĊĠĠĠĠĠĠĠ Ġ// Ġk oin Ġ= ĠK oin Application . create (). modules ( common ) ĊĊĠĠĠĠĠĠĠ Ġadmin Client Ġ= ĠKafka Admin Client . create ( test Props ) ĊĊĠĠĠĠĠĠĠ Ġadmin Client ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. create Topics ( mutable ListOf ( New Topic ( topic Name , Ġ 1 , Ġ 1 ))) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. all () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. get ( 2 0 , ĠTimeUnit . SECONDS ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ After All ĊĠĠĠ Ġinternal Ġfun ĠtearDown () Ġ{ ĊĠĠĠĠĠĠĠ Ġadmin Client . delete Topics ( mutable ListOf ( topic Name )) ĊĠĠĠĠĠĠĠ Ġadmin Client . close () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Experimental Std lib Api ĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġinternal Ġfun Ġtest L iveness Check () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġclient Ġ= ĠV ed t aks meld ing Client ( test Props , Ġtopic Name ) ĊĊĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ Ġclient . run L iveness Check () Ġ} ĊĊĠĠĠĠĠĠĠ Ġclient . stop () ĊĊĠĠĠĠĠĠĠ ĠassertThat Exception OfType ( Exception :: class . java ). is Thrown By Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ Ġclient . get Messages To Process () Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ ĠassertThat Exception OfType ( Exception :: class . java ). is Thrown By Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġrun Blocking Ġ{ Ġclient . run L iveness Check () Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Experimental Std lib Api ĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġget Messages () Ġ{ ĊĊĠĠĠĠĠĠĠ Ġval Ġclient Ġ= ĠV ed t aks meld ing Client ( test Props , Ġtopic Name ) ĊĠĠĠĠĠĠĠ Ġval Ġno Messages Expected Ġ= Ġclient . get Messages To Process () ĊĊĠĠĠĠĠĠĠ ĠassertThat ( no Messages Expected ). isEmpty () ĊĊĠĠĠĠĠĠĠ Ġval Ġproducer Ġ= ĠKafka Producer < String , ĠString >( test Props , ĠString Serializer (), ĠString Serializer ()) ĊĠĠĠĠĠĠĠ Ġval Ġgenerated Message Ġ= Ġgenerator . next () ĊĠĠĠĠĠĠĠ Ġproducer . send ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠProducer Record ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtopic Name , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ 0 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġgenerated Message . key , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġgenerated Message . message Body , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠlistOf ( Record Header (" type ", ĠSp le is Mel ding stype . V ed tak . name . toByteArray ())) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ). get ( 1 0 , ĠTimeUnit . SECONDS ) ĊĊĠĠĠĠĠĠĠ Ġval Ġone Message Expected Ġ= Ġclient . get Messages To Process () ĊĠĠĠĠĠĠĠ ĠassertThat ( one Message Expected ). has Size ( 1 ) ĊĊĠĠĠĠĠĠĠ Ġclient . confirm Processing Done () ĊĊĠĠĠĠĠĠĠ Ġval Ġzero Messages Expected Ġ= Ġclient . get Messages To Process () ĊĠĠĠĠĠĠĠ ĠassertThat ( zero Messages Expected ). isEmpty () ĊĊĠĠĠĠĠĠĠ Ġclient . stop () ĊĠĠĠ Ġ} Ċ } Ċ