package Ġno . nav . hel se . cli Ċ Ċ import Ġno . nav . hel se . cli . operations . get Offsets Ċ import Ġno . nav . hel se . cli . operations . get Partitions Ċ import Ġno . nav . rap ids _ and _ river s . cli . Consumer Producer Factory Ċ import Ġorg . apache . kafka . clients . admin . Admin Client Ċ import Ġorg . apache . kafka . clients . consumer . Offset And Metadata Ċ import Ġorg . apache . kafka . common . Topic Partition Ċ import Ġjava . time . LocalDateTime Ċ import Ġkotlin . system . exit Process Ċ Ċ internal Ġclass ĠSet Offsets Command Ġ: ĠCommand Ġ{ ĊĠĠĠ Ġoverride Ġval Ġname Ġ= Ġ" set _ offsets " ĊĊĠĠĠ Ġoverride Ġfun Ġusage () Ġ{ ĊĠĠĠĠĠĠĠ Ġprintln (" Usage : Ġ$ name Ġ< consumer Ġgroup > Ġ< topic >") ĊĠĠĠĠĠĠĠ Ġprintln (" Man ually Ġsets Ġoffsets Ġfor Ġa Ġconsumer Ġgroup ") ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġexecute ( factory : ĠConsumer Producer Factory , Ġargs : ĠList < String >) Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( args . size Ġ< Ġ 2 ) Ġthrow ĠRuntimeException (" Missing Ġrequired Ġconsumer Group Ġor Ġtopic Ġarg ") ĊĠĠĠĠĠĠĠ Ġval Ġconsumer Group Ġ= Ġargs [ 0 ] ĊĠĠĠĠĠĠĠ Ġval Ġtopic Ġ= Ġargs [ 1 ] ĊĠĠĠĠĠĠĠ Ġval Ġclient Ġ= Ġfactory . admin Client () ĊĊĠĠĠĠĠĠĠ Ġval Ġpartitions Ġ= Ġget Partitions ( client , Ġtopic ). sorted By Ġ{ Ġit . partition () Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġcurrent Offsets Ġ= Ġget Offsets ( client , ĠlistOf ( consumer Group )) ĊĊĠĠĠĠĠĠĠ Ġprintln ("$ topic Ġconsists Ġof Ġ${ partitions . size } Ġpartitions .") ĊĠĠĠĠĠĠĠ Ġprintln (" Enter Ġa Ġnew Ġoffset Ġfor Ġeach Ġpartition , Ġor Ġleave Ġblank Ġto Ġleave Ġun alter ed ") ĊĊĠĠĠĠĠĠĠ Ġval Ġoffsets Ġ= Ġpartitions . map NotNull Ġ{ Ġpartition Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprint (" Enter Ġa Ġnew Ġoffset Ġfor Ġpartition # ${ partition . partition ()} Ġ( current Ġis : Ġ${ current Offsets [ consumer Group ]?. get ( partition )?. offset ()} ): Ġ") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġread ln (). take Unless Ġ{ Ġit . isBlank () Ġ} ?. to Long OrNull ()?. let Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpartition Ġto Ġit ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġif Ġ( offsets . isEmpty ()) Ġreturn Ġprintln (" No Ġchanges ") ĊĊĠĠĠĠĠĠĠ Ġprintln (" You Ġhave Ġentered :") ĊĠĠĠĠĠĠĠ Ġoffsets . forEach Ġ{ Ġ( partition , Ġoffset ) Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġprintln ("\ t Partition Ġ# ${ partition . partition ()} : Ġ$ offset ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġprintln (" Is Ġthis Ġcorrect ? Ġ[ y / N ]") ĊĠĠĠĠĠĠĠ Ġval Ġanswer Ġ= Ġread ln () ĊĊĠĠĠĠĠĠĠ Ġif Ġ( answer . lowercase () Ġ!= Ġ" y ") Ġreturn Ġprintln (" Abort ing ") ĊĊĠĠĠĠĠĠĠ Ġprintln (" Setting Ġoffsets ") ĊĠĠĠĠĠĠĠ Ġclient . alter Consumer Group Offsets ( consumer Group , Ġoffsets . associate Ġ{ Ġ( partition , Ġoffset ) Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpartition Ġto ĠOffset And Metadata ( offset , Ġ" Offset Ġset Ġmanually Ġvia Ġcli Ġat Ġ${ LocalDateTime . now () }") ĊĠĠĠĠĠĠĠ Ġ}). all (). get () ĊĊĠĠĠĠĠĠĠ Ġprintln (" Current Ġoffsets :") ĊĠĠĠĠĠĠĠ Ġget Offsets ( client , ĠlistOf ( consumer Group )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. getValue ( consumer Group ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. forEach Ġ{ Ġ( partition , Ġoffset And Metadata ) Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġprintln ("# ${ partition . partition ()} : Ġ${ offset And Metadata . offset ()} ${ offset And Metadata . metadata ()?. take Unless Ġ{ Ġit . isBlank () Ġ} ?. let Ġ{ Ġ" Ġ($ it )" Ġ} Ġ?: Ġ"" }") ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġverify ( factory : ĠConsumer Producer Factory ) Ġ{} Ċ } Ċ