/** Ċ Ġ* ĠCopyright Ġ 2 0 2 0 . Ċ Ġ* Ċ Ġ* ĠLicensed Ġunder Ġthe ĠApache ĠLicense , ĠVersion Ġ 2 . 0 Ġ( the Ġ" License "); Ċ Ġ* Ġyou Ġmay Ġnot Ġuse Ġthis Ġfile Ġexcept Ġin Ġcompliance Ġwith Ġthe ĠLicense . Ċ Ġ* ĠYou Ġmay Ġobtain Ġa Ġcopy Ġof Ġthe ĠLicense Ġat Ċ Ġ* Ċ Ġ* ĠĠĠĠ Ġhttp :// www . apache . org / licenses / LICENSE - 2 . 0 Ċ Ġ* Ċ Ġ* ĠUnless Ġrequired Ġby Ġapplicable Ġlaw Ġor Ġagreed Ġto Ġin Ġwriting , Ġsoftware Ċ Ġ* Ġdistributed Ġunder Ġthe ĠLicense Ġis Ġdistributed Ġon Ġan Ġ" AS ĠIS " ĠBASIS , Ċ Ġ* ĠWITHOUT ĠWARRANTIES ĠOR ĠCONDITIONS ĠOF ĠANY ĠKIND , Ġeither Ġexpress Ġor Ġimplied . Ċ Ġ* ĠSee Ġthe ĠLicense Ġfor Ġthe Ġspecific Ġlanguage Ġgoverning Ġpermissions Ġand Ċ Ġ* Ġlimitations Ġunder Ġthe ĠLicense . Ċ Ġ*/ Ċ package Ġay ans en . programming . kafka . experiments Ċ Ċ import Ġay ans en . programming . avro . Sample Event Ċ import Ġay ans en . programming . kafka . experiments . Fixtures . generate Sample Events Ċ import Ġay ans en . programming . kafka . experiments . Fixtures . get Consumer Properties Ċ import Ġay ans en . programming . kafka . experiments . Fixtures . get Producer Properties Ċ import Ġorg . junit . jupiter . api . Test Ċ import Ġkotlin . test . assertEquals Ċ Ċ class ĠEvent Ordering IT Ġ{ ĊĊĠĠĠ Ġprivate Ġval Ġapp Consumer Topic : ĠString Ġ= Ġ" test _ topic _ v 1 " ĊĠĠĠ Ġprivate Ġval Ġapp Producer Topic : ĠString Ġ= Ġ" test _ topic _ mir rored _ v 1 " ĊĠĠĠ Ġprivate Ġval Ġkafka Mirror Ġ= ĠKafka Mirror ( ĊĠĠĠĠĠĠĠ Ġget Consumer Properties (), ĊĠĠĠĠĠĠĠ Ġget Producer Properties (), ĊĠĠĠĠĠĠĠ Ġapp Consumer Topic , ĊĠĠĠĠĠĠĠ Ġapp Producer Topic ĊĠĠĠ Ġ) ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠThis Ġtest Ġis Ġto Ġvalidate Ġthat Ġkafka Ġrecords Ġare Ġin Ġorder Ġfor Ġa Ġparticular Ġpartition ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġ` test Ġordering Ġof Ġevents Ġfor Ġa Ġparticular Ġpartition Ġassigned Ġbased Ġon Ġrecord Ġkey `() Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġfirst Sample Ġ= Ġgenerate Sample Events ( 1 0 , Ġ" first Sample ") ĊĠĠĠĠĠĠĠ Ġval Ġsecond Sample Ġ= Ġgenerate Sample Events ( 1 0 , Ġ" second Sample ") ĊĠĠĠĠĠĠĠ Ġkafka Mirror . process Records ( 5 0 0 , Ġ 2 ) ĊĠĠĠĠĠĠĠ ĠIntegration TestUtils . produce S ynchronously ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfalse , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġapp Consumer Topic , ĊĠĠĠĠĠĠĠĠĠĠĠĠ 0 , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfirst Sample ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ ĠIntegration TestUtils . produce S ynchronously ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġfalse , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġapp Consumer Topic , ĊĠĠĠĠĠĠĠĠĠĠĠĠ 1 , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġsecond Sample ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġval Ġconsumed Records Ġ= ĊĠĠĠĠĠĠĠĠĠĠĠ ĠIntegration TestUtils . wait Until Min Records Received < String , ĠSample Event >( app Producer Topic , Ġ 2 0 , Ġ 2 0 0 0 0 ) ĊĊĠĠĠĠĠĠĠ Ġval Ġconsumed First Sample Events Ġ= Ġconsumed Records ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġit . value (). eventName . contains (" first Sample ") Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. sorted By Ġ{ Ġit . timestamp () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġit . value () Ġ} ĊĊĠĠĠĠĠĠĠ Ġval Ġconsumed Second Sample Events Ġ= Ġconsumed Records ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġit . value (). eventName . contains (" second Sample ") Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. sorted By Ġ{ Ġit . timestamp () Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġit . value () Ġ} ĊĠĠĠĠĠĠĠ ĠassertEquals ( consumed First Sample Events , Ġfirst Sample . map Ġ{ Ġit . value Ġ}) ĊĠĠĠĠĠĠĠ ĠassertEquals ( consumed Second Sample Events , Ġsecond Sample . map Ġ{ Ġit . value Ġ}) ĊĠĠĠ Ġ} Ċ }