package Ġcom . fly er alarm . k af k amp . commands Ċ Ċ import Ġcom . fly er alarm . k af k amp . Mixed Value Ċ import Ġcom . fly er alarm . k af k amp . Record Ċ import Ġio . mock k . every Ċ import Ġio . mock k . just Ċ import Ġio . mock k . mock k Ċ import Ġio . mock k . runs Ċ import Ġio . mock k . verify Ċ import Ġio . mock k . verify Order Ċ import Ġorg . apache . kafka . clients . consumer . Consumer Records Ċ import Ġorg . apache . kafka . clients . consumer . Kafka Consumer Ċ import Ġorg . apache . kafka . common . Topic Partition Ċ import Ġorg . slf 4 j . Logger Ċ import Ġjava . time . Duration Ċ import Ġkotlin . test . Test Ċ import Ġkotlin . test . assertEquals Ċ Ċ class ĠPrint Test Ġ{ ĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` sub scri bes Ġto Ġconsumer Ġand Ġpol ls Ġuntil Ġno Ġmore Ġrecords Ġare Ġfound `() Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġconsumer Ġ= Ġmock k < Kafka Consumer < Mixed Value ?, ĠM ixed Value ? >>() ĊĊĠĠĠĠĠĠĠ Ġevery Ġ{ Ġconsumer . subscribe ( any < Collection < String >> ()) Ġ} Ġjust Ġruns ĊĠĠĠĠĠĠĠ Ġevery Ġ{ Ġconsumer . commit Sync () Ġ} Ġjust Ġruns ĊĠĠĠĠĠĠĠ Ġevery Ġ{ Ġconsumer . close () Ġ} Ġjust Ġruns ĊĊĠĠĠĠĠĠĠ Ġevery Ġ{ Ġconsumer . poll ( any < Duration >()) Ġ} Ġreturns Many ĠlistOf ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠConsumer Records ( map Of ( Topic Partition (" test ", Ġ 0 ) Ġto ĠlistOf ( Record (" test ", Ġ 0 , Ġ 0 , Ġmock k (), Ġmock k ()))) ), ĊĠĠĠĠĠĠĠĠĠĠĠ ĠConsumer Records ( map Of ( Topic Partition (" test ", Ġ 0 ) Ġto ĠlistOf ( Record (" test ", Ġ 0 , Ġ 0 , Ġmock k (), Ġmock k ()))) ), ĊĠĠĠĠĠĠĠĠĠĠĠ ĠConsumer Records ( empty Map ()) ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġval Ġcommand Ġ= ĠPrint ( mock k ( relax ed Ġ= Ġtrue ), Ġconsumer , Ġfalse ) ĊĠĠĠĠĠĠĠ Ġcommand . topic Ġ= Ġ" test " ĊĊĠĠĠĠĠĠĠ ĠassertEquals ( 0 , Ġcommand . call ()) ĊĊĠĠĠĠĠĠĠ Ġverify Order Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer . subscribe ( listOf (" test ")) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer . poll ( any < Duration >()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer . commit Sync () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer . poll ( any < Duration >()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer . commit Sync () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer . poll ( any < Duration >()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer . commit Sync () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer . close () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` does Ġnot Ġcommit Ġoffsets Ġif Ġrequested `() Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġconsumer Ġ= Ġmock k < Kafka Consumer < Mixed Value ?, ĠM ixed Value ? >>() ĊĊĠĠĠĠĠĠĠ Ġevery Ġ{ Ġconsumer . subscribe ( any < Collection < String >> ()) Ġ} Ġjust Ġruns ĊĠĠĠĠĠĠĠ Ġevery Ġ{ Ġconsumer . commit Sync () Ġ} Ġjust Ġruns ĊĠĠĠĠĠĠĠ Ġevery Ġ{ Ġconsumer . close () Ġ} Ġjust Ġruns ĊĊĠĠĠĠĠĠĠ Ġevery Ġ{ Ġconsumer . poll ( any < Duration >()) Ġ} Ġreturns Many ĠlistOf ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠConsumer Records ( map Of ( Topic Partition (" test ", Ġ 0 ) Ġto ĠlistOf ( Record (" test ", Ġ 0 , Ġ 0 , Ġmock k (), Ġmock k ()))) ), ĊĠĠĠĠĠĠĠĠĠĠĠ ĠConsumer Records ( map Of ( Topic Partition (" test ", Ġ 0 ) Ġto ĠlistOf ( Record (" test ", Ġ 0 , Ġ 0 , Ġmock k (), Ġmock k ()))) ), ĊĠĠĠĠĠĠĠĠĠĠĠ ĠConsumer Records ( empty Map ()) ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġval Ġcommand Ġ= ĠPrint ( mock k ( relax ed Ġ= Ġtrue ), Ġconsumer , Ġtrue ) ĊĠĠĠĠĠĠĠ Ġcommand . topic Ġ= Ġ" test " ĊĊĠĠĠĠĠĠĠ ĠassertEquals ( 0 , Ġcommand . call ()) ĊĊĠĠĠĠĠĠĠ Ġverify Order Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer . subscribe ( listOf (" test ")) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer . poll ( any < Duration >()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer . poll ( any < Duration >()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer . poll ( any < Duration >()) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer . close () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` logs Ġpretty - print ed Ġrecords `() Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġconsumer Ġ= Ġmock k < Kafka Consumer < Mixed Value ?, ĠM ixed Value ?> >( relax ed Ġ= Ġtrue ) ĊĊĠĠĠĠĠĠĠ Ġevery Ġ{ Ġconsumer . poll ( any < Duration >()) Ġ} Ġreturns Many ĠlistOf ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠConsumer Records ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmap Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠTopic Partition (" test ", Ġ 0 ) Ġto ĠlistOf ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠRecord ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" test ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ 0 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ 0 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠM ixed Value ( byte ArrayOf (), Ġ" KEY 1 "), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠM ixed Value ( byte ArrayOf (), Ġ" VALUE 1 ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ), ĊĠĠĠĠĠĠĠĠĠĠĠ ĠConsumer Records ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmap Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠTopic Partition (" test ", Ġ 1 ) Ġto ĠlistOf ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠRecord ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" test ", Ġ 1 , Ġ 1 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠM ixed Value ( byte ArrayOf (), Ġ" KEY 2 "), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠM ixed Value ( byte ArrayOf (), Ġ" VALUE 2 ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ), ĊĠĠĠĠĠĠĠĠĠĠĠ ĠConsumer Records ( empty Map ()) ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġval Ġlogger Ġ= Ġmock k < Logger >( relax ed Ġ= Ġtrue ) ĊĊĠĠĠĠĠĠĠ Ġval Ġcommand Ġ= ĠPrint ( logger , Ġconsumer , Ġfalse ) ĊĠĠĠĠĠĠĠ Ġcommand . topic Ġ= Ġ" test " ĊĊĠĠĠĠĠĠĠ ĠassertEquals ( 0 , Ġcommand . call ()) ĊĊĠĠĠĠĠĠĠ Ġverify Order Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ""" ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠRecord Ġat Ġoffset Ġ# 0 Ġin Ġtopic Ġ' test ' Ġ( Partition Ġ# 0 ): ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠKey : ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠKEY 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠValue : ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠVALUE 1 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ""" . trim Indent () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ""" ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠRecord Ġat Ġoffset Ġ# 1 Ġin Ġtopic Ġ' test ' Ġ( Partition Ġ# 1 ): ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠKey : ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠKEY 2 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠValue : ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠVALUE 2 ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ""" . trim Indent () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` logs Ġnumber Ġof Ġsuccessfully Ġprinted Ġrecords `() Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġconsumer Ġ= Ġmock k < Kafka Consumer < Mixed Value ?, ĠM ixed Value ?> >( relax ed Ġ= Ġtrue ) ĊĊĠĠĠĠĠĠĠ Ġevery Ġ{ Ġconsumer . poll ( any < Duration >()) Ġ} Ġreturns Many ĠlistOf ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠConsumer Records ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmap Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠTopic Partition (" test ", Ġ 0 ) Ġto ĠlistOf ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠRecord ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" test ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ 0 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ 0 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠM ixed Value ( byte ArrayOf (), Ġ" KEY 1 "), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠM ixed Value ( byte ArrayOf (), Ġ" VALUE 1 ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ), ĊĠĠĠĠĠĠĠĠĠĠĠ ĠConsumer Records ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmap Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠTopic Partition (" test ", Ġ 1 ) Ġto ĠlistOf ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠRecord ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" test ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ 1 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ 1 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠM ixed Value ( byte ArrayOf (), Ġ" KEY 2 "), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠM ixed Value ( byte ArrayOf (), Ġ" VALUE 2 ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ), ĊĠĠĠĠĠĠĠĠĠĠĠ ĠConsumer Records ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmap Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠTopic Partition (" test ", Ġ 1 ) Ġto ĠlistOf ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠRecord ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" test ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ 1 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ 1 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠM ixed Value ( byte ArrayOf (), Ġ" KEY 2 "), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠM ixed Value ( byte ArrayOf (), Ġ" VALUE 2 ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ), ĊĠĠĠĠĠĠĠĠĠĠĠ ĠConsumer Records ( empty Map ()) ĊĠĠĠĠĠĠĠ Ġ) ĊĊĠĠĠĠĠĠĠ Ġval Ġlogger Ġ= Ġmock k < Logger >( relax ed Ġ= Ġtrue ) ĊĊĠĠĠĠĠĠĠ Ġval Ġcommand Ġ= ĠPrint ( logger , Ġconsumer , Ġfalse ) ĊĠĠĠĠĠĠĠ Ġcommand . topic Ġ= Ġ" test " ĊĊĠĠĠĠĠĠĠ ĠassertEquals ( 0 , Ġcommand . call ()) ĊĊĠĠĠĠĠĠĠ Ġverify ( exact ly Ġ= Ġ 1 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" Finished Ġprinting Ġ 3 Ġrecord ( s ) Ġfrom Ġ' test '") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` ex its Ġwith Ġcode Ġ 1 Ġif Ġconsumers Ġthrows Ġexception Ġand Ġlogs Ġtotal Ġprinted `() Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġconsumer Ġ= Ġmock k < Kafka Consumer < Mixed Value ?, ĠM ixed Value ?> >( relax ed Ġ= Ġtrue ) ĊĠĠĠĠĠĠĠ Ġval Ġexception Ġ= ĠRuntimeException (" Test ") ĊĊĠĠĠĠĠĠĠ Ġevery Ġ{ Ġconsumer . poll ( any < Duration >()) Ġ} Ġreturns ĠConsumer Records ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmap Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠTopic Partition (" test ", Ġ 0 ) Ġto ĠlistOf ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠRecord ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" test ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ 0 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ 0 , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠM ixed Value ( byte ArrayOf (), Ġ" KEY 1 "), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠM ixed Value ( byte ArrayOf (), Ġ" VALUE 1 ") ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ) Ġand Then Throws Ġexception ĊĊĠĠĠĠĠĠĠ Ġval Ġlogger Ġ= Ġmock k < Logger >( relax ed Ġ= Ġtrue ) ĊĊĠĠĠĠĠĠĠ Ġval Ġcommand Ġ= ĠPrint ( logger , Ġconsumer , Ġfalse ) ĊĠĠĠĠĠĠĠ Ġcommand . topic Ġ= Ġ" test " ĊĊĠĠĠĠĠĠĠ ĠassertEquals ( 1 , Ġcommand . call ()) ĊĊĠĠĠĠĠĠĠ Ġverify Order Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . error (" Failed Ġto Ġprint Ġall Ġrecords Ġfrom Ġtopic Ġ' test '", Ġexception ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" 1 Ġrecord ( s ) Ġwere Ġprinted ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} Ċ } Ċ