package Ġcom . sk sam uel . co hort . kafka Ċ Ċ import Ġcom . sk sam uel . co hort . HealthCheck Ċ import Ġcom . sk sam uel . co hort . HealthCheck Result Ċ import Ġorg . apache . kafka . clients . consumer . Kafka Consumer Ċ import Ġkotlin . math . round ToInt Ċ Ċ /** Ċ Ġ* ĠA Ġ[ HealthCheck ] Ġthat Ġchecks Ġthat Ġa Ġkafka Ġconsumer Ġis Ġconsuming Ġa Ġminimum Ġnumber Ġof Ġmessages . Ċ Ġ* Ċ Ġ* ĠThis Ġcheck Ġcan Ġbe Ġuseful Ġto Ġdetect Ġst alled Ġconsumers . Ċ Ġ* Ċ Ġ* ĠThis Ġcheck Ġreports Ġhealthy Ġif Ġthe Ġrecords - consumed - rate Ġis Ġ>= Ġ[ min Receive Rate ]. Ċ Ġ*/ Ċ class ĠKafka Consumer Records Cons umed Rate HealthCheck ( ĊĠ Ġprivate Ġval Ġconsumer : ĠKafka Consumer < *, Ġ* >, ĊĠ Ġprivate Ġval Ġmin Receive Rate : ĠInt , Ċ ) Ġ: ĠHealth Check Ġ{ ĊĊĠ Ġprivate Ġval Ġmetric Name Ġ= Ġ" records - consumed - rate " ĊĊĠ Ġoverride Ġsuspend Ġfun Ġcheck (): ĠHealth Check Result Ġ{ ĊĠĠĠ Ġval Ġmetric Ġ= Ġconsumer . metrics (). values . first OrNull Ġ{ Ġit . metric Name (). name () Ġ== Ġmetric Name Ġ} ĊĠĠĠĠĠ Ġ?: Ġreturn ĠHealth Check Result . Un healthy (" Could Ġnot Ġlocate Ġkafka Ġmetric Ġ'${ metric Name }' ", Ġnull ) ĊĠĠĠ Ġval Ġsend Rate Ġ= Ġmetric . metric Value (). toString (). toDouble OrNull ()?. round ToInt () Ġ?: Ġ 0 ĊĠĠĠ Ġval Ġmsg Ġ= Ġ" Kafka Ġconsumer Ġ$ metric Name Ġ$ send Rate Ġ[ min Ġthreshold Ġ$ min Receive Rate ]" ĊĠĠĠ Ġreturn Ġif Ġ( send Rate Ġ< Ġmin Receive Rate ) ĊĠĠĠĠĠ ĠHealth Check Result . Un healthy ( msg , Ġnull ) ĊĠĠĠ Ġelse ĊĠĠĠĠĠ ĠHealth Check Result . Health y ( msg ) ĊĠ Ġ} Ċ } Ċ