import Ġcom . fasterxml . jackson . databind . ObjectMapper Ċ import Ġorg . apache . kafka . clients . consumer . Consumer Record Ċ import Ġorg . apache . kafka . clients . consumer . Mock Consumer Ċ import Ġorg . apache . kafka . clients . consumer . Offset Reset Strategy Ċ import Ġorg . apache . kafka . clients . producer . Mock Producer Ċ import Ġorg . apache . kafka . common . Topic Partition Ċ import Ġorg . apache . kafka . common . serialization . String Serializer Ċ import Ġorg . junit . Test Ċ import Ġru . ot us . op inion . kafka . Kafka Config Ċ import Ġru . ot us . op inion . kafka . Kafka Controller Ċ import Ġru . ot us . op inion . openapi . transport . models . Create Question Response Ċ import Ġru . ot us . op inion . openapi . transport . models . Dis crimin able Ċ import Ġru . ot us . op inion . openapi . transport . models . Response Ċ import Ġru . ot us . op inion . openapi . transport . models . Result Ċ import Ġru . ot us . op inion . openapi . transport . models . stubs . Stub s Ċ import Ġru . ot us . op inion . services . Question Service Ċ import Ġjava . util .* Ċ import Ġkotlin . test . assertEquals Ċ import Ġkotlin . test . assertTrue Ċ Ċ class ĠKafka Controller Test Ġ{ ĊĠĠĠ Ġprivate Ġval Ġom Ġ= ĠObjectMapper () ĊĠĠĠ Ġprivate Ġfun ĠDis crimin able . toJson (): ĠString Ġ= Ġom . writeValue AsString ( this ) ĊĠĠĠ Ġprivate Ġinline Ġfun Ġ< re ified ĠT : ĠDis crimin able > ĠString . fromJson () Ġ= Ġom . readValue < T >( this , ĠT :: class . java ) ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġrun Kafka () Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġconsumer Ġ= ĠMock Consumer < String , ĠString >( Offset Reset Strategy . EAR LI EST ) ĊĠĠĠĠĠĠĠ Ġval Ġproducer Ġ= ĠMock Producer ( true , ĠString Serializer (), ĠString Serializer ()) ĊĠĠĠĠĠĠĠ Ġval Ġconfig Ġ= ĠKafka Config ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġkafka Consumer Ġ= Ġconsumer , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġkafka Producer Ġ= Ġproducer , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġservice Ġ= ĠQuestion Service . getTest Service () ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġval Ġapp Ġ= ĠKafka Controller ( config ) ĊĠĠĠĠĠĠĠ Ġval Ġrequest Ġ= ĠStub s . create Request A ĊĠĠĠĠĠĠĠ Ġconsumer . schedule Poll Task Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer . re balance ( Collections . singletonList ( Topic Partition ( config . kafka Topic In , Ġ 0 ))) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer . add Record ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠConsumer Record ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconfig . kafka Topic In , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠPART ITION , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ 0 L , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" test - 1 ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġrequest . toJson () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġapp . stop () ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġstart Offsets : ĠMutable Map < Topic Partition , ĠLong > Ġ= Ġmutable Map Of () ĊĠĠĠĠĠĠĠ Ġval Ġtp Ġ= ĠTopic Partition ( config . kafka Topic In , ĠPART ITION ) ĊĠĠĠĠĠĠĠ Ġstart Offsets [ tp ] Ġ= Ġ 0 L ĊĠĠĠĠĠĠĠ Ġconsumer . update Beginning Offsets ( start Offsets ) ĊĊĠĠĠĠĠĠĠ Ġapp . run () ĊĊĠĠĠĠĠĠĠ Ġval Ġmessage Ġ= Ġproducer . history (). first () ĊĠĠĠĠĠĠĠ Ġval Ġresponse Ġ= Ġmessage . value (). fromJson < Response >() ĊĠĠĠĠĠĠĠ ĠassertTrue Ġ{ Ġresponse Ġis ĠCreate Question Response Ġ} ĊĠĠĠĠĠĠĠ ĠassertEquals ( request . requestId , Ġresponse . requestId ) ĊĠĠĠĠĠĠĠ ĠassertEquals ( Result . SUCCESS , Ġresponse . result ) ĊĠĠĠĠĠĠĠ ĠassertEquals ( 0 , Ġresponse . errors ?. size ) ĊĠĠĠĠĠĠĠ ĠassertEquals ( request . question , Ġ( response Ġas ĠCreate Question Response ). question ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġconst Ġval ĠPART ITION Ġ= Ġ 0 ĊĠĠĠ Ġ} Ċ }