@ file : Suppress (" NOTH ING _ TO _ INLINE ", Ġ" unused ") Ċ Ċ package Ġru . in for ion . lab 4 0 3 . common . kafka Ċ Ċ import Ġorg . apache . kafka . clients . admin . Admin Client Ċ import Ġorg . apache . kafka . clients . admin . Admin ClientConfig Ċ import Ġorg . apache . kafka . clients . admin . Offset Spec Ċ import Ġorg . apache . kafka . clients . admin . Topic Description Ċ import Ġorg . apache . kafka . common . Node Ċ import Ġorg . apache . kafka . common . Topic Partition Ċ import Ġru . in for ion . lab 4 0 3 . common . extensions . associate Ċ import Ġru . in for ion . lab 4 0 3 . common . extensions . if NotNull Ċ import Ġru . in for ion . lab 4 0 3 . common . logging . logger Ċ import Ġjava . io . Closeable ĊĊ Ċ class ĠKafka Admin Tool Ġconstructor ( val Ġbro kers : ĠString , Ġval Ġtimeout : ĠLong ) Ġ: ĠClose able Ġ{ ĊĠĠĠ Ġcompanion Ġobject Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġlog Ġ= Ġlogger () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġval Ġclient Ġ= ĠAdmin Client . create ( ĊĠĠĠĠĠĠĠ Ġmap Of ( ĊĠĠĠĠĠĠĠĠĠĠĠ ĠAdmin ClientConfig . BOOT STR AP _ SERVER S _ CONFIG Ġto Ġbro kers ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ) ĊĊĠĠĠ Ġfun Ġnodes (): ĠCollection < Node > Ġ= Ġclient . describe Cluster (). nodes (). get OrThrow ( timeout ) ĊĊĠĠĠ Ġfun Ġcluster Id (): ĠString Ġ= Ġclient . describe Cluster (). cluster Id (). get OrThrow ( timeout ) ĊĊĠĠĠ Ġfun Ġcheck (): ĠBoolean Ġ= Ġclient . describe Cluster (). cluster Id (). get OrNull ( timeout ) Ġ!= Ġnull ĊĊĠĠĠ Ġfun Ġlist Topics (): ĠSet < String > Ġ= Ġclient . list Topics (). names (). get OrThrow ( timeout ) ĊĊĠĠĠ Ġfun Ġdescribe Topics ( topics : ĠCollection < String >): ĠCollection < Topic Description > Ġ= ĊĠĠĠĠĠĠĠ Ġclient . describe Topics ( topics ). all (). get OrThrow ( timeout ). values ĊĊĠĠĠ Ġfun Ġlist Offsets ( offsets : ĠCollection < Topic Partition >): ĠMap < Topic Partition , ĠLong > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġparameters Ġ= Ġoffsets . associate With Ġ{ ĠOffset Spec . latest () Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġresult Ġ= Ġclient . list Offsets ( parameters ). all (). get OrThrow ( timeout ) ĊĠĠĠĠĠĠĠ Ġreturn Ġresult . associate Ġ{ Ġit . key Ġto Ġit . value . offset () Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġlist Consumer Group Offsets ( group : ĠString ): ĠMap < Topic Partition , ĠLong > Ġ= ĊĠĠĠĠĠĠĠ Ġclient . list Consumer Group Offsets ( group ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. partitions To Offset And Metadata () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. get OrThrow ( timeout ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. associate Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġit . key Ġto Ġit . value . offset () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġtopics Info ( topics : ĠCollection < String >) Ġ= Ġdescribe Topics ( topics ). map Ġ{ Ġtopic Ġ-> ĊĠĠĠĠĠĠĠ Ġval Ġparameters Ġ= Ġtopic . to Topic Partitions () ĊĠĠĠĠĠĠĠ Ġval Ġpartitions Ġ= Ġlist Offsets ( parameters ). map Ġ{ Ġ( partition , Ġoffset ) Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠ ĠPartition Info ( partition . partition (), Ġ- 1 , Ġoffset ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ ĠTopic Info ( topic . name (), Ġpartitions ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġconsumers Info ( groups : ĠCollection < String >) Ġ= Ġgroups . associate With Ġ{ Ġgroup Ġ-> ĊĠĠĠĠĠĠĠ Ġval Ġoffsets Ġ= Ġlist Consumer Group Offsets ( group ) ĊĊĠĠĠĠĠĠĠ Ġ// Ġassume Ġgroup Ġfor Ġsingle Ġtopic ĊĠĠĠĠĠĠĠ Ġoffsets . keys . first OrNull () Ġif NotNull Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġpartitions Ġ= Ġlist Offsets ( offsets . keys ). map Ġ{ Ġ( partition , Ġsize ) Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġoffset Ġ= Ġoffsets . getValue ( partition ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠPartition Info ( partition . partition (), Ġoffset , Ġsize ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ ĠTopic Info ( topic (), Ġpartitions ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġoverride Ġfun Ġclose () Ġ= Ġclient . close () Ċ }