package Ġcom . info bip . k af k ist ry . service . topic Ċ Ċ import Ġorg . apache . kafka . clients . admin . Config Entry Ċ import Ġcom . info bip . k af k ist ry . kafka . Config Value Ċ import Ġcom . info bip . k af k ist ry . kafka . Existing Config Ċ import Ġcom . info bip . k af k ist ry . service . Config Value Change Ċ import Ġcom . info bip . k af k ist ry . service . Value Inspection Ċ import Ġorg . springframework . stereotype . Component Ċ Ċ @ Component Ċ class ĠConfig Value Inspector Ġ{ ĊĊĠĠĠ Ġfun Ġcheck Config Property ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġname Key : ĠString , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġactual Value : ĠConfig Value , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpected Value : ĠString ?, ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcluster Config : ĠExisting Config ĊĠĠĠ Ġ): ĠValue Inspection Ġ= Ġinspect ( name Key , Ġactual Value , Ġexpected Value , Ġcluster Config ) ĊĊĠĠĠ Ġfun Ġneed Include Config Value ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġname Key : ĠString , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġactual Value : ĠConfig Value , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcluster Config : ĠExisting Config ĊĠĠĠ Ġ): ĠBoolean Ġ= Ġinspect ( name Key , Ġactual Value , Ġnull , Ġcluster Config ). valid . not () ĊĊĠĠĠ Ġfun Ġrequired Config Value Change ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġname Key : ĠString , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġactual Value : ĠConfig Value , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpected Value : ĠString ?, ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcluster Config : ĠExisting Config ĊĠĠĠ Ġ): ĠConfig Value Change ? Ġ= Ġinspect ( name Key , Ġactual Value , Ġexpected Value , Ġcluster Config ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. take If Ġ{ Ġit . current Value Ġ!= Ġit . expected Value Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ? . let Ġ{ ĠConfig Value Change ( name Key , Ġit . current Value , Ġit . expected Value , Ġit . expect ing Cluster Default ) Ġ} ĊĊĠĠĠ Ġfun Ġis Value Same As Expected By Cluster ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġname Key : ĠString , Ġvalue : ĠString ?, Ġcluster Config : ĠExisting Config ĊĠĠĠ Ġ): ĠBoolean Ġ= Ġcluster Config . extract Cluster Expected Value ( name Key )?. value Ġ== Ġvalue ĊĊĠĠĠ Ġfun Ġcluster DefaultValue ( cluster Config : ĠExisting Config , Ġname Key : ĠString ): ĠConfig Value ? Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġcluster Config . extract Cluster Expected Value ( name Key ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġinspect ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġname Key : ĠString , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġactual Value : ĠConfig Value , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġexpected Value : ĠString ?, ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcluster Config : ĠExisting Config ĊĠĠĠ Ġ): ĠValue Inspection Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġif Ġ( expected Value Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġinspect Registry Undefined Value ( cluster Config , Ġname Key , Ġactual Value ) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġinspect Registry Expected Value ( expected Value , Ġactual Value ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġinspect Registry Expected Value ( expected Value : ĠString , Ġactual Value : ĠConfig Value ): ĠValue Inspection Ġ{ ĊĠĠĠĠĠĠĠ Ġ// since Ġvalue Ġis Ġdefined Ġin Ġregistry Ġexplicitly , Ġactual Ġvalue Ġneed Ġto Ġbe Ġsame Ġas Ġit ĊĠĠĠĠĠĠĠ Ġreturn Ġif Ġ( expected Value Ġ== Ġactual Value . value ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠValue Inspection ( true , Ġactual Value . value , Ġexpected Value , Ġfalse ) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠValue Inspection ( false , Ġactual Value . value , Ġexpected Value , Ġfalse ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġinspect Registry Undefined Value ( cluster Config : ĠExisting Config , Ġname Key : ĠString , Ġactual Value : ĠConfig Value ): ĠValue Inspection Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġcluster Expected Value Ġ= Ġcluster Config . extract Cluster Expected Value ( name Key ) ĊĠĠĠĠĠĠĠ Ġreturn Ġif Ġ( cluster Expected Value Ġ== Ġnull ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( actual Value . default ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// value Ġis Ġnot Ġspecified Ġin Ġregistry 's Ġconfig Ġand Ġvalue Ġon Ġcluster Ġis Ġdefault , Ġso Ġok Ġthen ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠValue Inspection ( true , Ġactual Value . value , Ġactual Value . value , Ġtrue ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠValue Inspection ( false , Ġactual Value . value , Ġnull , Ġtrue ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( cluster Expected Value . value Ġ== Ġactual Value . value ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠValue Inspection ( true , Ġactual Value . value , Ġcluster Expected Value . value , Ġtrue ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( name Key ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" message . format . version " Ġ-> Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġif Ġ( actual Value . value . matches Message Format Version ( cluster Expected Value . value )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠValue Inspection ( true , Ġactual Value . value , Ġactual Value . value , Ġtrue ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠValue Inspection ( false , Ġactual Value . value , Ġcluster Expected Value . value , Ġtrue ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> ĠValue Inspection ( false , Ġactual Value . value , Ġcluster Expected Value . value , Ġtrue ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun ĠExisting Config . extract Cluster Expected Value ( name Key : ĠString ): ĠConfig Value ? Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġcluster Config Ġ= Ġthis ĊĠĠĠĠĠĠĠ Ġreturn Ġnothing < Config Value >() ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. orElse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ// handle Ġall Ġexception ally Ġnamed Ġcluster Ġserver Ġproperties ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( name Key ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" delete . retention . ms " Ġ-> Ġcluster Config [" log . clean er . delete . retention . ms "] ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" file . delete . delay . ms " Ġ-> Ġcluster Config [" log . segment . delete . delay . ms "] ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" flush . messages " Ġ-> Ġcluster Config [" log . flush . interval . messages "] ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" flush . ms " Ġ-> Ġcluster Config [" log . flush . interval . ms "] ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" max . comp action . lag . ms " Ġ-> Ġcluster Config [" log . clean er . max . comp action . lag . ms "] ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" min . comp action . lag . ms " Ġ-> Ġcluster Config [" log . clean er . min . comp action . lag . ms "] ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" max . message . bytes " Ġ-> Ġcluster Config [" message . max . bytes "] ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" min . clean able . dirty . ratio " Ġ-> Ġcluster Config [" log . clean er . min . clean able . ratio "] ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" message . format . version " Ġ-> Ġcluster Config [" log . message . format . version "] ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" segment . index . bytes " Ġ-> Ġcluster Config [" log . index . size . max . bytes "] ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" segment . j itter . ms " Ġ-> Ġcluster Config [" log . roll . j itter . ms "] ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" segment . ms " Ġ-> Ġcluster Config [" log . roll . ms "] ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" follow er . replication . th rott led . replicas " Ġ-> ĠConfig Value ( null , Ġdefault Ġ= Ġtrue , ĠreadOnly Ġ= Ġfalse , Ġsensitive Ġ= Ġfalse , Ġsource Ġ= ĠConfig Entry . Config Source . DEFAULT _ CONFIG ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" leader . replication . th rott led . replicas " Ġ-> ĠConfig Value ( null , Ġdefault Ġ= Ġtrue , ĠreadOnly Ġ= Ġfalse , Ġsensitive Ġ= Ġfalse , Ġsource Ġ= ĠConfig Entry . Config Source . DEFAULT _ CONFIG ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġnull ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. orElse Ġ{ Ġcluster Config [" log .$ name Key "] Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. orElse Ġ{ Ġcluster Config [ name Key ] Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ? . take If Ġ{ Ġit . value Ġ!= Ġnull Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġ< T > Ġnothing (): ĠT ? Ġ= Ġnull ĊĠĠĠ Ġprivate Ġfun Ġ< T > ĠT ?. orElse ( supplier : Ġ() Ġ-> ĠT ?): ĠT ? Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġthis Ġ?: Ġsupplier () ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun ĠString ?. matches Message Format Version ( expected : ĠString ?): ĠBoolean Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ( expected Ġ== Ġnull ) Ġreturn Ġfalse ĊĠĠĠĠĠĠĠ Ġreturn Ġthis ?. startsWith ( expected ) Ġ?: Ġfalse ĊĠĠĠ Ġ} Ċ Ċ }