Ċ import Ġkotlinx . coroutines . delay Ċ import Ġorg . apache . kafka . clients . consumer . Kafka Consumer Ċ import Ġorg . apache . kafka . common . Partition Info Ċ import Ġuk . gov . d wp . data works . logging . Data works Logger Ċ import Ġkotlin . time . Experimental Time Ċ import Ġkotlin . time . seconds Ċ Ċ object ĠSubscriber Utility Ġ{ ĊĊĠĠĠ Ġ@ Experimental Time ĊĠĠĠ Ġtail rec Ġsuspend Ġfun Ġ< T > Ġsubscribe ( consumer : ĠKafka Consumer < T , ĠT >, Ġincludes Regex : ĠRegex , Ġexclude s Regex : ĠRegex ? Ġ= Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġcurrent Subscription Ġ= Ġconsumer . subscription () ĊĊĠĠĠĠĠĠĠ Ġval Ġtopics Ġ= Ġ Ġwith ( included Topics ( consumer , Ġincludes Regex )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexclude s Regex ?. let Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfilter Not ( exclude s Regex :: matches ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġ?: Ġthis ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġif Ġ( topics . toSet () Ġ!= Ġcurrent Subscription Ġ&& Ġtopics . isNotEmpty ()) Ġ{ ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġtopics . minus ( current Subscription ). forEach Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġlogger . info (" New Ġtopic Ġfound ", Ġ" topic " Ġto Ġit ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠĠĠĠĠ Ġconsumer . subscribe ( topics ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġif Ġ( current Subscription . isNotEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn ĊĠĠĠĠĠĠĠ Ġ} ĊĊĠĠĠĠĠĠĠ Ġlogger . info (" No Ġtopics Ġand Ġno Ġcurrent Ġsubscription Ġtrying Ġagain ") ĊĠĠĠĠĠĠĠ Ġdelay ( 1 . seconds ) ĊĠĠĠĠĠĠĠ Ġsubscribe ( consumer , Ġincludes Regex , Ġexclude s Regex ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġ< T > Ġincluded Topics ( consumer : ĠKafka Consumer < T , ĠT >, Ġinclusion Regex : ĠRegex ): ĠList < String > Ġ= ĊĠĠĠĠĠĠĠ Ġconsumer . list Topics () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map ( Map . Entry < String , ĠList < Partition Info >>:: key ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter ( in clusion Regex :: matches ) ĊĊĠĠĠ Ġprivate Ġval Ġlogger Ġ= ĠData works Logger . getLogger ( Subscriber Utility :: class ) Ċ } Ċ