package Ġio . github . cf st out . k aco on . kafka . consumer Ċ Ċ import Ġorg . apache . kafka . clients . consumer . Consumer Record Ċ import Ġorg . apache . kafka . clients . consumer . Offset And Metadata Ċ import Ġorg . apache . kafka . common . Topic Partition Ċ import Ġorg . junit . jupiter . api . Assertions Ċ import Ġorg . junit . jupiter . api . Test Ċ Ċ class ĠOffset Tracking Test Ġ{ ĊĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġoffset Tracking Ġ= ĠOffset Tracking () ĊĠĠĠĠĠĠĠ Ġval Ġpartition Ġ= ĠTopic Partition (" foo ", Ġ 0 ) ĊĠĠĠĠĠĠĠ Ġval Ġrecord 0 Ġ= Ġrecord ( partition , Ġ 0 ) ĊĠĠĠĠĠĠĠ Ġval Ġrecord 1 Ġ= Ġrecord ( partition , Ġ 1 ) ĊĠĠĠĠĠĠĠ Ġval Ġrecord 2 Ġ= Ġrecord ( partition , Ġ 2 ) ĊĊĠĠĠĠĠĠĠ Ġprivate Ġfun Ġrecord ( topic Partition : ĠTopic Partition , Ġoffset : ĠLong ): ĠConsumer Record < *, Ġ*> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĠConsumer Record ( topic Partition . topic (), Ġtopic Partition . partition (), Ġoffset , Ġnull , Ġnull ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Back log Sizes () Ġ{ ĊĠĠĠĠĠĠĠ Ġoffset Tracking . started ( record 0 ) ĊĠĠĠĠĠĠĠ Ġoffset Tracking . started ( record 1 ) ĊĊĠĠĠĠĠĠĠ Ġvar Ġback log Sizes : ĠMap < Topic Partition , ĠInt > Ġ= Ġoffset Tracking . back log Sizes () ĊĠĠĠĠĠĠĠ Ġvar Ġexpected Ġ= Ġmap Of ( partition Ġto Ġ 2 ) ĊĠĠĠĠĠĠĠ ĠAssertions . assertEquals ( expected , Ġback log Sizes ) ĊĊĠĠĠĠĠĠĠ Ġoffset Tracking . complete ( record 1 ) ĊĠĠĠĠĠĠĠ Ġback log Sizes Ġ= Ġoffset Tracking . back log Sizes () ĊĠĠĠĠĠĠĠ Ġexpected Ġ= Ġmap Of ( partition Ġto Ġ 1 ) ĊĠĠĠĠĠĠĠ ĠAssertions . assertEquals ( expected , Ġback log Sizes ) ĊĊĠĠĠĠĠĠĠ Ġoffset Tracking . complete ( record 0 ) ĊĠĠĠĠĠĠĠ Ġoffset Tracking . started ( record 2 ) ĊĠĠĠĠĠĠĠ Ġback log Sizes Ġ= Ġoffset Tracking . back log Sizes () ĊĠĠĠĠĠĠĠ Ġexpected Ġ= Ġmap Of ( partition Ġto Ġ 1 ) ĊĠĠĠĠĠĠĠ ĠAssertions . assertEquals ( expected , Ġback log Sizes ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ@ Test ĊĠĠĠ Ġfun Ġtest Offsets To Commit () Ġ{ ĊĠĠĠĠĠĠĠ Ġoffset Tracking . started ( record 0 ) ĊĠĠĠĠĠĠĠ Ġoffset Tracking . started ( record 1 ) ĊĊĠĠĠĠĠĠĠ Ġvar Ġpair : ĠPair < Topic Partition , ĠOffset And Metadata > Ġ= Ġoffset Tracking . offsets To Commit ( partition , ĠOffset And Metadata ( 0 )) ĊĠĠĠĠĠĠĠ Ġ// ĠNothing Ġcompleted Ġyet ĊĠĠĠĠĠĠĠ ĠAssertions . assertEquals ( 0 , Ġpair . second . offset ()) ĊĊĠĠĠĠĠĠĠ Ġoffset Tracking . complete ( record 1 ) ĊĠĠĠĠĠĠĠ Ġpair Ġ= Ġoffset Tracking . offsets To Commit ( partition , ĠOffset And Metadata ( 0 )) ĊĠĠĠĠĠĠĠ Ġ// Ġthe Ġsecond Ġrecord Ġwas Ġcompleted , Ġbut Ġthat Ġmeans Ġwe Ġstill Ġneed Ġto Ġprocess Ġthe Ġfirst Ġso ĊĠĠĠĠĠĠĠ Ġ// Ġbasically Ġthe Ġsame Ġas Ġnothing Ġbeing Ġcompleted ĊĠĠĠĠĠĠĠ ĠAssertions . assertEquals ( 0 , Ġpair . second . offset ()) ĊĊĠĠĠĠĠĠĠ Ġoffset Tracking . complete ( record 0 ) ĊĠĠĠĠĠĠĠ Ġpair Ġ= Ġoffset Tracking . offsets To Commit ( partition , ĠOffset And Metadata ( 0 )) ĊĠĠĠĠĠĠĠ Ġ// ĠNow Ġwe Ġcomplete Ġthe Ġfirst Ġso Ġposition Ġshould Ġbe Ġthe Ġrecord Ġafter Ġrecord Ġ" 1 " Ġ( e . g . Ġ 2 ) ĊĠĠĠĠĠĠĠ ĠAssertions . assertEquals ( 2 , Ġpair . second . offset ()) ĊĠĠĠ Ġ} Ċ }