package Ġfund . cyber . common . kafka Ċ Ċ import Ġfund . cyber . search . model . chains . Chain EntityType Ċ import Ġfund . cyber . search . model . chains . Chain Info Ċ import Ġorg . apache . kafka . clients . consumer . Consumer Ċ import Ġorg . apache . kafka . common . Topic Partition Ċ Ċ fun ĠChain EntityType . kafka Topic Name ( chain Info : ĠChain Info ) Ġ Ġ= Ġ"${ chain Info . name }_${ this . name }_ PU MP " Ċ Ċ fun ĠConsumer < Any , ĠAny >. read Topic Last Offset ( topic : ĠString ): ĠLong Ġ{ ĊĠĠĠ Ġval Ġpartitions Ġ= Ġpartitions For ( topic ) ĊĠĠĠ Ġif Ġ( partitions Ġ== Ġnull Ġ|| Ġpartitions . size Ġ== Ġ 0 ) Ġreturn Ġ- 1 L ĊĠĠĠ Ġif Ġ( partitions . size Ġ> Ġ 1 ) Ġthrow ĠRuntimeException (" Topic Ġhave Ġmore Ġthan Ġone Ġpartition ") ĊĊĠĠĠ Ġval Ġpartition Ġ= ĠTopic Partition ( partitions . first (). topic (), Ġpartitions . first (). partition ()) ĊĠĠĠ Ġassign ( listOf ( partition )) ĊĠĠĠ Ġseek ToEnd ( listOf ( partition )) ĊĠĠĠ Ġreturn Ġposition ( partition ) Ġ- Ġ 1 Ċ } Ċ