package Ġcom . info bip . k af k ist ry . service . topic . validation . rules Ċ Ċ import Ġcom . info bip . k af k ist ry . k af k ast ate . Kafka Clusters State Provider Ċ import Ġcom . info bip . k af k ist ry . service . k af k ast ream s . K Stream AppId Ċ import Ġcom . info bip . k af k ist ry . service . k af k ast ream s . K Streams Apps Provider Ċ import Ġcom . info bip . k af k ist ry . service . k af k ast ream s . Topic K Streams In vol vement Ċ import Ġcom . info bip . k af k ist ry . model . Kafka Cluster Identifier Ċ import Ġcom . info bip . k af k ist ry . model . Topic Description Ċ import Ġcom . info bip . k af k ist ry . model . Topic Name Ċ import Ġcom . info bip . k af k ist ry . service . K af k ist ry Integr ity Exception Ċ import Ġcom . info bip . k af k ist ry . service . properties For Cluster Ċ import Ġcom . info bip . k af k ist ry . service . with Cluster Properties Ċ import Ġcom . info bip . k af k ist ry . utils . Cluster Topic Filter Ċ import Ġcom . info bip . k af k ist ry . utils . Cluster Topic Filter Properties Ċ import Ġorg . springframework . boot . context . properties . Configuration Properties Ċ import Ġorg . springframework . boot . context . properties . Nested Configuration Property Ċ import Ġorg . springframework . context . annotation . Configuration Ċ import Ġorg . springframework . stereotype . Component Ċ Ċ @ Configuration Ċ @ Configuration Properties (" app . topic - validation . k stream - partition - count ") Ċ class ĠK Stream Partition Count Rule Properties Ġ{ ĊĊĠĠĠ Ġ@ Nested Configuration Property ĊĠĠĠ Ġvar Ġenabled On Ġ= ĠCluster Topic Filter Properties () Ċ } ĊĊ Ċ @ Component Ċ class ĠK Stream Partition Count Rule ( ĊĠĠĠ Ġprivate Ġval Ġk Streams Apps Provider : ĠK Streams Apps Provider , ĊĠĠĠ Ġprivate Ġval Ġcluster State Provider : ĠKafka Clusters State Provider , ĊĠĠĠ Ġproperties : ĠK Stream Partition Count Rule Properties , Ċ ) Ġ: ĠValidation Rule Ġ{ ĊĊĠĠĠ Ġprivate Ġval Ġfilter Ġ= ĠCluster Topic Filter ( properties . enabled On ) ĊĊĠĠĠ Ġoverride Ġfun Ġcheck ( topic Description View : ĠTopic Description View , Ġcluster Metadata : ĠCluster Metadata ): ĠRule Violation ? Ġ{ ĊĠĠĠĠĠĠĠ Ġif Ġ(! filter ( cluster Metadata . ref . identifier , Ġtopic Description View . name )) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġvalid () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġmiss Matching App Topics Ġ= Ġmiss Matching Topics ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtopic Description View . name , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtopic Description View . properties . partition Count , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcluster Metadata . ref . identifier , ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġif Ġ( miss Matching App Topics . isEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġvalid () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġvi olated ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġmessage Ġ= Ġ" Topic 's Ġpartition Ġcount Ġof Ġ% partition Count % Ġviol ates ĠK Stream Ġco - partition ing Ġwith Ġfollowing Ġ" Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" (${ miss Matching App Topics . size }) Ġother Ġtopics Ġinvolved Ġwithin Ġthe Ġsame ĠK Stream Ġapplication : Ġ\ n " Ġ+ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmiss Matching App Topics . map Indexed Ġ{ Ġindex , Ġ_ Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ"% topic $ index % [% partitions $ index %] Ġ@ Ġ% k Stream App $ index %" ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ}. join ToString (", Ġ"), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġplaceholders Ġ= Ġmiss Matching App Topics ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Indexed Ġ{ Ġindex , Ġk Stream Topic Partitions Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠlistOf ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" k Stream App $ index " Ġto ĠPlace holder (" k Stream - app ", Ġk Stream Topic Partitions . k Stream AppId ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" topic $ index " Ġto ĠPlace holder (" topic - name ", Ġk Stream Topic Partitions . topic ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" partitions $ index " Ġto ĠPlace holder (" partition - count ", Ġk Stream Topic Partitions . partitions ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. toMap () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. plus ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" partition Count " Ġto ĠPlace holder (" partition - count ", Ġtopic Description View . properties . partition Count ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ), ĊĠĠĠĠĠĠĠĠĠĠĠ Ġseverity Ġ= ĠRule Violation . Severity . WARNING , ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġmiss Matching Topics ( ĊĠĠĠĠĠĠĠ Ġtopic Name : ĠTopic Name , ĊĠĠĠĠĠĠĠ Ġpartition Count : ĠInt , ĊĠĠĠĠĠĠĠ Ġcluster Identifier : ĠKafka Cluster Identifier , ĊĠĠĠ Ġ): ĠList < K Stream Topic Partitions > Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġk Stream Apps In vol vement Ġ= Ġk Streams Apps Provider . topic K Stream Apps In vol vement ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcluster Identifier Ġ= Ġcluster Identifier , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtopic Name Ġ= Ġtopic Name , ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġif Ġ( k Stream Apps In vol vement Ġ== ĠTopic K Streams In vol vement . NONE ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġempty List () ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġ( k Stream Apps In vol vement . input Of Ġ+ ĠlistOf NotNull ( k Stream Apps In vol vement . internal In )) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Ġ{ Ġapp Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġapp . kafka Stream AppId Ġto Ġ( app . input Topics Ġ+ Ġapp . k Stream Internal Topics ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġit Ġ!= Ġtopic Name Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map NotNull Ġ{ Ġtopic Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpartition Count Of ( cluster Identifier , Ġtopic )?. let Ġ{ Ġtopic Ġto Ġit Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Ġ{ Ġ(_, Ġpartitions ) Ġ-> Ġpartitions Ġ!= Ġpartition Count Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. toMap () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. flatMap Ġ{ Ġ( appId , Ġmiss Matches ) Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġmiss Matches . map Ġ{ Ġ( topic , Ġpartitions ) Ġ-> ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠK Stream Topic Partitions ( appId , Ġtopic , Ġpartitions ) ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġdata Ġclass ĠK Stream Topic Partitions ( ĊĠĠĠĠĠĠĠ Ġval Ġk Stream AppId : ĠK Stream AppId , ĊĠĠĠĠĠĠĠ Ġval Ġtopic : ĠTopic Name , ĊĠĠĠĠĠĠĠ Ġval Ġpartitions : ĠInt , ĊĠĠĠ Ġ) ĊĊĠĠĠ Ġprivate Ġfun Ġpartition Count Of ( ĊĠĠĠĠĠĠĠ Ġcluster Identifier : ĠKafka Cluster Identifier , Ġtopic Name : ĠTopic Name ĊĠĠĠ Ġ): ĠInt ? Ġ= Ġcluster State Provider . get Latest Cluster State ( cluster Identifier ) ĊĠĠĠĠĠĠĠ Ġ. value OrNull () ĊĠĠĠĠĠĠĠ Ġ? . topics ?. find Ġ{ Ġit . name Ġ== Ġtopic Name Ġ} ĊĠĠĠĠĠĠĠ Ġ? . partitions Assignments ?. size ĊĊĠĠĠ Ġoverride Ġfun Ġdo Fix Config ( topic Description : ĠTopic Description , Ġcluster Metadata : ĠCluster Metadata ): ĠTopic Description Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġtopic Properties Ġ= Ġtopic Description . properties For Cluster ( cluster Metadata . ref ) ĊĠĠĠĠĠĠĠ Ġval Ġpartition Count Ġ= Ġtopic Properties . partition Count ĊĠĠĠĠĠĠĠ Ġval Ġmiss Matching Topics Ġ= Ġmiss Matching Topics ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġtopic Name Ġ= Ġtopic Description . name , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġpartition Count Ġ= Ġpartition Count , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcluster Identifier Ġ= Ġcluster Metadata . ref . identifier , ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġif Ġ( miss Matching Topics . isEmpty ()) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġreturn Ġtopic Description ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġpartition Count Topics Ġ= Ġmiss Matching Topics . groupBy ({ Ġit . partitions Ġ}, Ġ{ Ġit . topic Ġ}) ĊĠĠĠĠĠĠĠ Ġif Ġ( partition Count Topics . size Ġ> Ġ 2 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠK af k ist ry Integr ity Exception ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" Imp ossible Ġto Ġfix Ġpartition Ġcount Ġdue Ġto Ġconflicting Ġpartition Ġcounts Ġof Ġother Ġtopics : Ġ$ partition Count Topics " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġresolved Partition Count Ġ= Ġpartition Count Topics . keys . first () ĊĠĠĠĠĠĠĠ Ġif Ġ( resolved Partition Count Ġ< Ġpartition Count ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠK af k ist ry Integr ity Exception ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" Can 't Ġfix Ġpartition Ġcount Ġbecause Ġpartition Ġcount Ġcant Ġbe Ġreduced Ġfrom Ġ$ partition Count Ġto Ġ$ resolved Partition Count " ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġtopic Description . with Cluster Properties ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcluster Metadata . ref . identifier , Ġtopic Properties . copy ( partition Count Ġ= Ġresolved Partition Count ) ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ} Ċ }