package Ġnl . mit che ln ij dam . k af k are try test . kafka Ċ Ċ import Ġorg . apache . kafka . clients . consumer . Consumer Ċ import Ġorg . apache . kafka . clients . consumer . Consumer Records Ċ import Ġorg . apache . kafka . common . Topic Partition Ċ Ċ /** Ċ Ġ* ĠCustom Ġseek Ġfunction Ġwhich Ġwill Ġseek Ġto Ġthe Ġnext Ġ( max Ġ+ Ġ 1 ) Ġoffset Ġfor Ġeach Ġ[ Topic Partition ] Ġin Ġthe Ġ[ Consumer Records ]. Ċ Ġ* Ċ Ġ* Ġ@ author ĠMit che l ĠN ij dam Ċ Ġ*/ Ċ fun Ġ< K , ĠV > ĠConsumer Records < K , ĠV >. seek To Next ( consumer : ĠConsumer < *, Ġ* >) Ġ{ ĊĠĠĠ Ġthis . toList () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. groupBy Ġ{ ĠTopic Partition ( it . topic (), Ġit . partition ()) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Values Ġ{ Ġ(_, Ġvalue ) Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvalue . max By Ġ{ Ġit . offset () Ġ} !!. offset () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}. forEach Ġ{ Ġ( topic Partition , Ġmax Offset ) Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer . seek ( topic Partition , Ġmax Offset Ġ+ Ġ 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ċ } Ċ Ċ /** Ċ Ġ* ĠCustom Ġseek Ġfunction Ġwhich Ġwill Ġseek Ġto Ġthe Ġfirst / sm allest Ġoffset Ġfor Ġeach Ġ[ Topic Partition ] Ġin Ġall Ġ[ Consumer Records ]. Ċ Ġ*/ Ċ fun Ġ< K , ĠV > ĠConsumer Records < K , ĠV >. seek To Current ( consumer : ĠConsumer < *, Ġ* >) Ġ{ ĊĠĠĠ Ġthis . toList () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. groupBy Ġ{ ĠTopic Partition ( it . topic (), Ġit . partition ()) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Values Ġ{ Ġ(_, Ġvalue ) Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġvalue . min By Ġ{ Ġit . offset () Ġ} !!. offset () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ}. forEach Ġ{ Ġ( topic Partition , Ġmin Offset ) Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer . seek ( topic Partition , Ġmin Offset ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ċ }