Refactor CC capacity configuration classes

cluster-operator/src/main/java/io/strimzi/operator/cluster/model/CruiseControl.java

@@ -35,7 +35,7 @@
 import io.strimzi.api.kafka.model.kafka.cruisecontrol.CruiseControlTemplate;
 import io.strimzi.certs.CertAndKey;
 import io.strimzi.operator.cluster.ClusterOperatorConfig;
-import io.strimzi.operator.cluster.model.cruisecontrol.Capacity;
+import io.strimzi.operator.cluster.model.cruisecontrol.CapacityConfiguration;
 import io.strimzi.operator.cluster.model.cruisecontrol.CruiseControlConfiguration;
 import io.strimzi.operator.cluster.model.cruisecontrol.HashLoginServiceApiCredentials;
 import io.strimzi.operator.cluster.model.logging.LoggingModel;
@@ -111,7 +111,7 @@ public class CruiseControl extends AbstractModel implements SupportsMetrics, Sup
     private boolean authEnabled;
     private HashLoginServiceApiCredentials apiCredentials;
     @SuppressFBWarnings({"UWF_FIELD_NOT_INITIALIZED_IN_CONSTRUCTOR"}) // This field is initialized in the fromCrd method
-    protected Capacity capacity;
+    private CapacityConfiguration capacityConfiguration;
     @SuppressFBWarnings({"UWF_FIELD_NOT_INITIALIZED_IN_CONSTRUCTOR"}) // This field is initialized in the fromCrd method
     private MetricsModel metrics;
     private LoggingModel logging;
@@ -197,6 +197,7 @@ public static CruiseControl fromCrd(
             result.image = image;
 
             KafkaConfiguration kafkaConfiguration = new KafkaConfiguration(reconciliation, kafkaClusterSpec.getConfig().entrySet());
+            result.capacityConfiguration = new CapacityConfiguration(reconciliation, ccSpec, kafkaBrokerNodes, kafkaStorage, kafkaBrokerResources);
             result.updateConfigurationWithDefaults(ccSpec, kafkaConfiguration);
 
             CruiseControlConfiguration ccConfiguration = result.configuration;
@@ -207,7 +208,6 @@ public static CruiseControl fromCrd(
 
             // To avoid illegal storage configurations provided by the user,
             // we rely on the storage configuration provided by the KafkaAssemblyOperator
-            result.capacity = new Capacity(reconciliation, kafkaCr.getSpec(), kafkaBrokerNodes, kafkaStorage, kafkaBrokerResources);
             result.readinessProbeOptions = ProbeUtils.extractReadinessProbeOptionsOrDefault(ccSpec, ProbeUtils.DEFAULT_HEALTHCHECK_OPTIONS);
             result.livenessProbeOptions = ProbeUtils.extractLivenessProbeOptionsOrDefault(ccSpec, ProbeUtils.DEFAULT_HEALTHCHECK_OPTIONS);
             result.gcLoggingEnabled = ccSpec.getJvmOptions() == null ? JvmOptions.DEFAULT_GC_LOGGING_ENABLED : ccSpec.getJvmOptions().isGcLoggingEnabled();
@@ -530,7 +530,7 @@ public LoggingModel logging()   {
     public ConfigMap generateConfigMap(MetricsAndLogging metricsAndLogging) {
         Map<String, String> configMapData = new HashMap<>();
         configMapData.put(SERVER_CONFIG_FILENAME, configuration.asOrderedProperties().asPairs());
-        configMapData.put(CAPACITY_CONFIG_FILENAME, capacity.toString());
+        configMapData.put(CAPACITY_CONFIG_FILENAME, capacityConfiguration.toJson());
         configMapData.putAll(ConfigMapUtils.generateMetricsAndLogConfigMapData(reconciliation, this, metricsAndLogging));
 
         return ConfigMapUtils