package Ġcom . info bip . k af k ist ry . service . topic . wizard Ċ Ċ import Ġcom . fasterxml . jackson . databind . ObjectMapper Ċ import Ġcom . fasterxml . jackson . module . kotlin . register Kotlin Module Ċ import Ġorg . apache . kafka . common . config . Topic Config .* Ċ import Ġcom . info bip . k af k ist ry . k af k ast ate . Kafka Cluster State Ċ import Ġcom . info bip . k af k ist ry . k af k ast ate . State Data Ċ import Ġcom . info bip . k af k ist ry . k af k ast ate . State Type Ċ import Ġcom . info bip . k af k ist ry . model .* Ċ import Ġcom . info bip . k af k ist ry . service . Topic Wizard Exception Ċ import Ġcom . info bip . k af k ist ry . service . generator . Overrides Min im izer Ċ import Ġcom . info bip . k af k ist ry . service . resources . Required Resources Inspector Ċ import Ġcom . info bip . k af k ist ry . service . resources . Topic Resource Required Usage s Ċ import Ġcom . info bip . k af k ist ry . service . topic . validation . Naming Validator Ċ import Ġcom . info bip . k af k ist ry . model . Cluster Ref Ċ import Ġcom . info bip . k af k ist ry . model . Kafka Cluster Identifier Ċ import Ġcom . info bip . k af k ist ry . model . Tag Ċ 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 Ċ import Ġjava . util .* Ċ import Ġkotlin . math . min Ċ Ċ @ Configuration Ċ @ Configuration Properties (" app . topic - wizard ") Ċ class ĠTopic Wizard Properties Ġ{ ĊĊĠĠĠ Ġvar Ġdefault Partition Count Ġ= Ġ 1 ĊĠĠĠ Ġvar Ġpartition Threshold s Ġ= Ġ"{} " ĊĊĠĠĠ Ġ@ Nested Configuration Property ĊĠĠĠ Ġvar Ġsegment Size Ġ= ĠSegment Size Properties () ĊĊĠĠĠ Ġclass ĠSegment Size Properties Ġ{ ĊĠĠĠĠĠĠĠ Ġvar Ġto Retention Ratio Small Ġ= Ġ 0 . 5 ĊĠĠĠĠĠĠĠ Ġvar Ġretention Size Big Threshold Ġ= Ġ 1 0 0 L Ġshl Ġ 2 0 ĊĠĠĠĠĠĠĠ Ġvar Ġto Retention Ratio Big Ġ= Ġ 0 . 1 ĊĠĠĠ Ġ} Ċ } Ċ Ċ @ Component Ċ class ĠTopic Wizard Config Generator ( ĊĠĠĠĠĠĠĠ Ġtopic Name Generator : ĠOptional < Topic Name Generator >, ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġminim izer : ĠOverrides Min im izer , ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġnaming Validator : ĠN aming Validator , ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġrequired Resources Inspector : ĠRequired Resources Inspector , ĊĠĠĠĠĠĠĠ Ġprivate Ġval Ġproperties : ĠTopic Wizard Properties , Ċ ) Ġ{ ĊĊĠĠĠ Ġprivate Ġval Ġname Generator : ĠTopic Name Generator Ġ= Ġtopic Name Generator . orElse ( Default Topic Name Generator ) ĊĠĠĠ Ġprivate Ġval Ġpartition Count Assign or Ġ= ĠPartition Count Assign or ( properties . default Partition Count , Ġproperties . partition Threshold s ) ĊĊĠĠĠ Ġfun Ġgenerate Topic Description ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwizard Answers : ĠTopic Creation Wizard Answers , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġall Clusters : ĠMap < Cluster Ref , ĠState Data < Kafka Cluster State >> ĊĠĠĠ Ġ): ĠTopic Description Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġtopic Presence Ġ= Ġwizard Answers . presence ĊĠĠĠĠĠĠĠ Ġval Ġcluster Configs Ġ= Ġall Clusters ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. filter Keys Ġ{ Ġtopic Presence . need ToBe On Cluster ( it ) Ġ} ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ. map Values Ġ{ Ġgenerate Topic Configuration For Cluster ( wizard Answers , Ġit . key , Ġit . value ) Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġtopic Description Ġ= ĠTopic Description ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġname Ġ= Ġgenerate Name ( wizard Answers . topic Name Metadata ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġproperties Ġ= Ġignored Properties (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġowner Ġ= Ġwizard Answers . team Name , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġproducer Ġ= Ġwizard Answers . producer ServiceName , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġdescription Ġ= Ġgenerate Description ( wizard Answers ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresource Requirements Ġ= Ġwizard Answers . resource Requirements , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġpresence Ġ= Ġtopic Presence , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconfig Ġ= Ġempty Map (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġper Cluster Properties Ġ= Ġcluster Configs . map Ġ{ Ġit . key . identifier Ġto Ġit . value . properties Ġ}. toMap (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġper Cluster Config Overrides Ġ= Ġcluster Configs . map Ġ{ Ġit . key . identifier Ġto Ġit . value . config Ġ}. toMap (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġper Tag Properties Ġ= Ġempty Map (), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġper Tag Config Overrides Ġ= Ġempty Map (), ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġreturn Ġminim izer . minimize Overrides ( topic Description , Ġcluster Configs . keys . toList ()) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġ/** ĊĠĠĠĠ Ġ* ĠThis Ġvalue Ġhere Ġwill Ġbe Ġcompletely Ġignored Ġbecause Ġwe Ġgenerate ĠTopic Properties Ġper Ġeach Ġcluster ĊĠĠĠĠ Ġ* Ġwhich Ġwill Ġbe Ġmoved Ġ" up / global " Ġto Ġreplace Ġthis Ġvalue . ĊĠĠĠĠ Ġ* ĊĠĠĠĠ Ġ* ĠBut Ġsomething Ġneeds Ġto Ġbe Ġspecified Ġhere Ġbecause Ġfield ĠTopic Description . properties Ġis Ġnon - nullable . ĊĠĠĠĠ Ġ*/ ĊĠĠĠ Ġinternal Ġfun Ġignored Properties () Ġ= ĠTopic Properties ( 1 , Ġ 1 ) ĊĊĠĠĠ Ġfun Ġgenerate Name ( topic Name Metadata : ĠTopic Name Metadata ): ĠString Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġname Generator . generate Topic Name ( topic Name Metadata ). also Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġnaming Validator . validate Topic Name ( it ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġgenerate Description ( wizard Answers : ĠTopic Creation Wizard Answers ): ĠString Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġwizard Answers . purpose Ġ+ Ġ"\ n \ n " Ġ+ Ġ" Topic Ġgenerated Ġby ĠK af k ist ry Ġwizard " ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġgenerate Topic Configuration For Cluster ( ĊĠĠĠĠĠĠĠ Ġwizard Answers : ĠTopic Creation Wizard Answers , ĊĠĠĠĠĠĠĠ Ġcluster Ref : ĠCluster Ref , ĊĠĠĠĠĠĠĠ Ġcluster State : ĠState Data < Kafka Cluster State > ĊĠĠĠ Ġ): ĠGenerated Config Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġpartition Count Ġ= Ġdetermine Partition Count ( wizard Answers . resource Requirements , Ġcluster Ref ) ĊĠĠĠĠĠĠĠ Ġval Ġ( replication Factor , Ġmin In Sync Replicas ) Ġ= Ġdetermine Topic Availability Gu arante es ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwizard Answers , Ġcluster Ref . identifier , Ġcluster State ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġval Ġtopic Properties Ġ= ĠTopic Properties ( partition Count , Ġreplication Factor ) ĊĠĠĠĠĠĠĠ Ġval Ġretention Ms Ġ= Ġwizard Answers . resource Requirements . retention . to Millis () ĊĠĠĠĠĠĠĠ Ġval Ġresource Usage s Ġ= Ġrequired Resources Inspector . inspect Topic Resources ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtopic Properties Ġ= Ġtopic Properties , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresource Requirements Ġ= Ġwizard Answers . resource Requirements , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcluster Ref Ġ= Ġcluster Ref , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcluster Info Ġ= Ġcluster State . value OrNull ()?. cluster Info ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġval Ġretention Bytes Ġ= Ġresource Usage s . disk Usage Per Partition Replica ĊĠĠĠĠĠĠĠ Ġval Ġsegment Bytes Ġ= Ġdetermine Segment Bytes ( retention Bytes ) ĊĠĠĠĠĠĠĠ Ġreturn ĠGenerated Config ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġproperties Ġ= Ġtopic Properties , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġconfig Ġ= Ġmap Of ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠRET ENT ION _ BYTES _ CONFIG Ġto Ġ"$ retention Bytes ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠRET ENT ION _ MS _ CONFIG Ġto Ġ"$ retention Ms ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠMIN _ IN _ SYNC _ REPL IC AS _ CONFIG Ġto Ġ"$ min In Sync Replicas ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ ĠSEG MENT _ BYTES _ CONFIG Ġto Ġ"$ segment Bytes ", ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ), ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġresource Required Usage s Ġ= Ġresource Usage s , ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġcomments Ġ= ĠlistOf () ĊĠĠĠĠĠĠĠ Ġ) ĊĠĠĠ Ġ} ĊĊĠĠ Ġfun Ġdetermine Segment Bytes ( retention Bytes : ĠLong ): ĠLong Ġ{ ĊĠĠĠĠĠĠĠ Ġreturn Ġwith ( properties . segment Size ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġval Ġthreshold Segment Bytes Ġ= Ġretention Size Big Threshold . times ( to Retention Ratio Small ). to Long () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġwhen Ġ( retention Bytes Ġ< Ġretention Size Big Threshold ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġtrue Ġ-> Ġretention Bytes . times ( to Retention Ratio Small ). to Long () ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġfalse Ġ-> Ġretention Bytes . times ( to Retention Ratio Big ). to Long (). coerce At Least ( threshold Segment Bytes ) ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġfun Ġdetermine Partition Count ( ĊĠĠĠĠĠĠĠĠĠĠĠ Ġresource Requirements : ĠResource Requirements , ĊĠĠĠĠĠĠĠĠĠĠĠ Ġcluster Ref : ĠCluster Ref ĊĠĠĠ Ġ): ĠInt Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġmessages Rate Ġ= Ġresource Requirements . messages Rate ( cluster Ref ) ĊĠĠĠĠĠĠĠ Ġif Ġ( messages Rate . amount Ġ<= Ġ 0 ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ Ġthrow ĠTopic Wizard Exception (" messages Ġrate Ġ( Through put ) Ġmust Ġbe Ġ> Ġ 0 . ĠCluster : Ġ${ cluster Ref . identifier }, ĠValue : Ġ$ messages Rate ") ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġreturn Ġpartition Count Assign or . assign Partition Count ( messages Rate . rate Per Sec (), Ġcluster Ref ) ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġfun Ġdetermine Topic Availability Gu arante es ( ĊĠĠĠĠĠĠĠ Ġwizard Answers : ĠTopic Creation Wizard Answers , ĊĠĠĠĠĠĠĠ Ġcluster Identifier : ĠKafka Cluster Identifier , ĊĠĠĠĠĠĠĠ Ġcluster State : ĠState Data < Kafka Cluster State > ĊĠĠĠ Ġ): ĠTopic Availability Gu arante es Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġ( wanted Replication Factor , Ġtol erate Data Loss ) Ġ= Ġwhen Ġ( wizard Answers . high Availability ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠHigh Availability . NONE Ġ-> Ġreturn ĠTopic Availability Gu arante es ( 1 , Ġ 1 ) ĊĠĠĠĠĠĠĠĠĠĠĠ ĠHigh Availability . BASIC Ġ-> Ġ 2 Ġto Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠ ĠHigh Availability . STRONG _ AVAIL ABILITY Ġ-> Ġ 3 Ġto Ġtrue ĊĠĠĠĠĠĠĠĠĠĠĠ ĠHigh Availability . STRONG _ D UR ABILITY Ġ-> Ġ 3 Ġto Ġfalse ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġnumber Broker Nodes Ġ= Ġwhen Ġ( cluster State . state Type ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠState Type . VISIBLE Ġ-> Ġcluster State . value (). cluster Info . node Ids . size ĊĠĠĠĠĠĠĠĠĠĠĠ ĠState Type . DISABLED Ġ-> Ġwanted Replication Factor ĊĠĠĠĠĠĠĠĠĠĠĠ Ġelse Ġ-> Ġthrow ĠTopic Wizard Exception ( ĊĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠĠ Ġ" Can 't Ġsuggest Ġreplication Ġfactor Ġfor Ġcluster Ġ'$ cluster Identifier ' Ġbecause Ġit 's Ġstate Ġis Ġ${ cluster State . state Type }" ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠĠĠĠĠ Ġval Ġreplication Factor Ġ= Ġmin ( wanted Replication Factor , Ġnumber Broker Nodes ) ĊĠĠĠĠĠĠĠ Ġreturn Ġif Ġ( tol erate Data Loss ) Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠTopic Availability Gu arante es ( replication Factor , Ġ 1 ) ĊĠĠĠĠĠĠĠ Ġ} Ġelse Ġ{ ĊĠĠĠĠĠĠĠĠĠĠĠ ĠTopic Availability Gu arante es ( replication Factor , Ġmin In Sync Replicas Ġ= Ġreplication Factor Ġ- Ġ 1 ) ĊĠĠĠĠĠĠĠ Ġ} ĊĠĠĠ Ġ} ĊĊĠĠĠ Ġprivate Ġdata Ġclass ĠGenerated Config ( ĊĠĠĠĠĠĠĠ Ġval Ġproperties : ĠTopic Properties , ĊĠĠĠĠĠĠĠ Ġval Ġconfig : ĠTopic ConfigMap , ĊĠĠĠĠĠĠĠ Ġval Ġresource Required Usage s : ĠTopic Resource Required Usage s , ĊĠĠĠĠĠĠĠ Ġval Ġcomments : ĠList < String > ĊĠĠĠ Ġ) Ċ Ċ } Ċ Ċ private Ġclass ĠPartition Count Assign or ( ĊĠĠĠ Ġval Ġdefault Partition Count : ĠInt , ĊĠĠĠ Ġwizard Partition Threshold s Json : ĠString Ċ ) Ġ{ ĊĊĠĠĠ Ġprivate Ġval Ġthresholds Config : ĠThreshold s Config Ġ= ĠObjectMapper () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. register Kotlin Module () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ. readValue ( wizard Partition Threshold s Json , ĠThreshold s Config :: class . java ) ĊĊĠĠĠ Ġfun Ġassign Partition Count ( msg Per Sec : ĠDouble , Ġcluster Ref : ĠCluster Ref ): ĠInt Ġ{ ĊĠĠĠĠĠĠĠ Ġval Ġthresholds Ġ= Ġthresholds Config . overrides [ cluster Ref . identifier ] ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ?: Ġcluster Ref . tags . map NotNull Ġ{ Ġthresholds Config . tag Overrides [ it ] Ġ}. first OrNull () ĊĠĠĠĠĠĠĠĠĠĠĠ Ġ?: Ġthresholds Config . default ĊĠĠĠĠĠĠĠ Ġreturn Ġthresholds . floor Entry ( msg Per Sec )?. value Ġ?: Ġdefault Partition Count ĊĠĠĠ Ġ} Ċ Ċ } Ċ Ċ private Ġdata Ġclass ĠThreshold s Config ( ĊĠĠĠ Ġval Ġdefault : ĠTreeMap < Double , ĠInt > Ġ= ĠTreeMap (), ĊĠĠĠ Ġval Ġoverrides : ĠMap < Kafka Cluster Identifier , ĠTreeMap < Double , ĠInt >> Ġ= Ġempty Map (), ĊĠĠĠ Ġval Ġtag Overrides : ĠMap < Tag , ĠTreeMap < Double , ĠInt >> Ġ= Ġempty Map (), Ċ ) Ċ Ċ private Ġdata Ġclass ĠTopic Availability Gu arante es ( val Ġreplication Factor : ĠInt , Ġval Ġmin In Sync Replicas : ĠInt )