Fix missing jmx reporter (#11379)

Signed-off-by: Paolo Patierno <ppatierno@live.com>

Author: ppatierno

CHANGELOG.md

@@ -51,6 +51,7 @@
   If you have any custom logging configuration, you might need to update it during the upgrade to Kafka 4.0.
 * Kubernetes events for Pod restarts no longer have the Pod as the `regardingObject`.
   If you are using `regardingObject` as a `field-selector` for listing events you must update the selector to specify the Kafka resource instead.
+* From Kafka 4.0.0, to enable the JMXReporter you must either enable metrics in `.spec.kafka.metrics`, or explicitly add JMXReporter in `metric.reporters`.
 
 ## 0.45.0
 

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

@@ -60,6 +60,8 @@ public class KafkaBrokerConfigurationBuilder {
     private final static String PLACEHOLDER_CERT_STORE_PASSWORD_CONFIG_PROVIDER_ENV_VAR = "${strimzienv:CERTS_STORE_PASSWORD}";
     private final static String PLACEHOLDER_OAUTH_CLIENT_SECRET_TEMPLATE_CONFIG_PROVIDER_ENV_VAR = "${strimzienv:STRIMZI_%s_OAUTH_CLIENT_SECRET}";
 
+    private final static String KAFKA_JMX_REPORTER_CLASS = "org.apache.kafka.common.metrics.JmxReporter";
+
     private final StringWriter stringWriter = new StringWriter();
     private final PrintWriter writer = new PrintWriter(stringWriter);
     private final Reconciliation reconciliation;
@@ -787,44 +789,52 @@ private void configProviders(KafkaConfiguration userConfig)    {
      *
      * @param userConfig                The User configuration - Kafka broker configuration options specified by the user in the Kafka custom resource
      * @param injectCcMetricsReporter   Inject the Cruise Control Metrics Reporter into the configuration
+     * @param isMetricsEnabled          Flag to indicate if metrics are enabled. If they are we inject the JmxReporter into the configuration
      *
      * @return Returns the builder instance
      */
-    public KafkaBrokerConfigurationBuilder withUserConfiguration(KafkaConfiguration userConfig, boolean injectCcMetricsReporter)  {
-        if (userConfig != null && !userConfig.getConfiguration().isEmpty()) {
-            // We have to create a copy of the configuration before we modify it
-            userConfig = new KafkaConfiguration(userConfig);
-
-            // Configure the configuration providers => we have to inject the Strimzi ones
-            configProviders(userConfig);
-
-            if (injectCcMetricsReporter)  {
-                // We configure the Cruise Control Metrics Reporter is needed
-                if (userConfig.getConfigOption(CruiseControlMetricsReporter.KAFKA_METRIC_REPORTERS_CONFIG_FIELD) != null) {
-                    if (!userConfig.getConfigOption(CruiseControlMetricsReporter.KAFKA_METRIC_REPORTERS_CONFIG_FIELD).contains(CruiseControlMetricsReporter.CRUISE_CONTROL_METRIC_REPORTER)) {
-                        userConfig.setConfigOption(CruiseControlMetricsReporter.KAFKA_METRIC_REPORTERS_CONFIG_FIELD, userConfig.getConfigOption(CruiseControlMetricsReporter.KAFKA_METRIC_REPORTERS_CONFIG_FIELD) + "," + CruiseControlMetricsReporter.CRUISE_CONTROL_METRIC_REPORTER);
-                    }
-                } else {
-                    userConfig.setConfigOption(CruiseControlMetricsReporter.KAFKA_METRIC_REPORTERS_CONFIG_FIELD, CruiseControlMetricsReporter.CRUISE_CONTROL_METRIC_REPORTER);
-                }
-            }
+    public KafkaBrokerConfigurationBuilder withUserConfiguration(KafkaConfiguration userConfig, boolean injectCcMetricsReporter, boolean isMetricsEnabled)  {
+        // We have to create a copy of the configuration before we modify it
+        userConfig = userConfig != null ? new KafkaConfiguration(userConfig) : new KafkaConfiguration(reconciliation, new ArrayList<>());
+
+        // Configure the configuration providers => we have to inject the Strimzi ones
+        configProviders(userConfig);
 
+        addMetricReporters(userConfig, injectCcMetricsReporter, isMetricsEnabled);
+
+        // print user config with Strimzi injections
+        if (!userConfig.getConfiguration().isEmpty()) {
             printSectionHeader("User provided configuration");
             writer.println(userConfig.getConfiguration());
             writer.println();
-        } else {
-            // Configure the configuration providers => we have to inject the Strimzi ones
-            configProviders(userConfig);
+        }
 
-            if (injectCcMetricsReporter) {
-                // There is no user provided configuration. But we still need to inject the Cruise Control Metrics Reporter
-                printSectionHeader("Cruise Control Metrics Reporter");
-                writer.println(CruiseControlMetricsReporter.KAFKA_METRIC_REPORTERS_CONFIG_FIELD + "=" + CruiseControlMetricsReporter.CRUISE_CONTROL_METRIC_REPORTER);
-                writer.println();
+        return this;
+    }
+
+    /**
+     * Add metric reporters to the corresponding Kafka configuration
+     *
+     * @param userConfig                The User configuration - Kafka broker configuration options specified by the user in the Kafka custom resource
+     * @param injectCcMetricsReporter   Inject the Cruise Control Metrics Reporter into the configuration
+     * @param injectJmxReporter         Inject the JMX Reporter into the configuration
+     */
+    private void addMetricReporters(KafkaConfiguration userConfig, boolean injectCcMetricsReporter, boolean injectJmxReporter) {
+        if (injectJmxReporter)  {
+            if (userConfig.getConfigOption("metric.reporters") != null && !userConfig.getConfigOption("metric.reporters").contains(KAFKA_JMX_REPORTER_CLASS)) {
+                userConfig.setConfigOption("metric.reporters", userConfig.getConfigOption("metric.reporters") + "," + KAFKA_JMX_REPORTER_CLASS);
+            } else {
+                userConfig.setConfigOption("metric.reporters", KAFKA_JMX_REPORTER_CLASS);
             }
         }
 
-        return this;
+        if (injectCcMetricsReporter)  {
+            if (userConfig.getConfigOption("metric.reporters") != null && !userConfig.getConfigOption("metric.reporters").contains(CruiseControlMetricsReporter.CRUISE_CONTROL_METRIC_REPORTER)) {
+                userConfig.setConfigOption("metric.reporters", userConfig.getConfigOption("metric.reporters") + "," + CruiseControlMetricsReporter.CRUISE_CONTROL_METRIC_REPORTER);
+            } else {
+                userConfig.setConfigOption("metric.reporters", CruiseControlMetricsReporter.CRUISE_CONTROL_METRIC_REPORTER);
+            }
+        }
     }
 
     /**

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

@@ -1814,7 +1814,7 @@ private String generatePerBrokerConfiguration(NodeRef node, KafkaPool pool, Map<
                 .withCruiseControl(cluster, ccMetricsReporter, node.broker())
                 .withTieredStorage(cluster, tieredStorage)
                 .withQuotas(cluster, quotas)
-                .withUserConfiguration(configuration, node.broker() && ccMetricsReporter != null)
+                .withUserConfiguration(configuration, node.broker() && ccMetricsReporter != null, metrics.isEnabled())
                 .build()
                 .trim();
     }

cluster-operator/src/test/java/io/strimzi/operator/cluster/model/KafkaBrokerConfigurationBuilderTest.java

@@ -457,7 +457,7 @@ public void testOpaAuthorizationWithTls() {
     @ParallelTest
     public void testNullUserConfiguration()  {
         String configuration = new KafkaBrokerConfigurationBuilder(Reconciliation.DUMMY_RECONCILIATION, NODE_REF)
-                .withUserConfiguration(null, false)
+                .withUserConfiguration(null, false, false)
                 .build();
 
         assertThat(configuration, isEquivalent("node.id=2",
@@ -473,7 +473,7 @@ public void testNullUserConfiguration()  {
     @ParallelTest
     public void testNullUserConfigurationAndCCReporter()  {
         String configuration = new KafkaBrokerConfigurationBuilder(Reconciliation.DUMMY_RECONCILIATION, NODE_REF)
-                .withUserConfiguration(null, true)
+                .withUserConfiguration(null, true, false)
                 .build();
 
         assertThat(configuration, isEquivalent("node.id=2",
@@ -487,13 +487,30 @@ public void testNullUserConfigurationAndCCReporter()  {
                 "metric.reporters=com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter"));
     }
 
+    @ParallelTest
+    public void testNullUserConfigurationAndMetricsEnabled()  {
+        String configuration = new KafkaBrokerConfigurationBuilder(Reconciliation.DUMMY_RECONCILIATION, NODE_REF)
+                .withUserConfiguration(null, false, true)
+                .build();
+
+        assertThat(configuration, isEquivalent("node.id=2",
+                "config.providers=strimzienv,strimzifile,strimzidir",
+                "config.providers.strimzienv.class=org.apache.kafka.common.config.provider.EnvVarConfigProvider",
+                "config.providers.strimzienv.param.allowlist.pattern=.*",
+                "config.providers.strimzifile.class=org.apache.kafka.common.config.provider.FileConfigProvider",
+                "config.providers.strimzifile.param.allowed.paths=/opt/kafka",
+                "config.providers.strimzidir.class=org.apache.kafka.common.config.provider.DirectoryConfigProvider",
+                "config.providers.strimzidir.param.allowed.paths=/opt/kafka",
+                "metric.reporters=org.apache.kafka.common.metrics.JmxReporter"));
+    }
+
     @ParallelTest
     public void testEmptyUserConfiguration()  {
         Map<String, Object> userConfiguration = new HashMap<>();
         KafkaConfiguration kafkaConfiguration = new KafkaConfiguration(Reconciliation.DUMMY_RECONCILIATION, userConfiguration.entrySet());
 
         String configuration = new KafkaBrokerConfigurationBuilder(Reconciliation.DUMMY_RECONCILIATION, NODE_REF)
-                .withUserConfiguration(kafkaConfiguration, false)
+                .withUserConfiguration(kafkaConfiguration, false, false)
                 .build();
 
         assertThat(configuration, isEquivalent("node.id=2",
@@ -517,7 +534,7 @@ public void testUserConfiguration()  {
         KafkaConfiguration kafkaConfiguration = new KafkaConfiguration(Reconciliation.DUMMY_RECONCILIATION, userConfiguration.entrySet());
 
         String configuration = new KafkaBrokerConfigurationBuilder(Reconciliation.DUMMY_RECONCILIATION, NODE_REF)
-                .withUserConfiguration(kafkaConfiguration, false)
+                .withUserConfiguration(kafkaConfiguration, false, false)
                 .build();
 
         assertThat(configuration, isEquivalent("node.id=2",
@@ -544,7 +561,7 @@ public void testUserConfigurationWithConfigProviders()  {
 
         // Broker
         String configuration = new KafkaBrokerConfigurationBuilder(Reconciliation.DUMMY_RECONCILIATION, NODE_REF)
-                .withUserConfiguration(kafkaConfiguration, false)
+                .withUserConfiguration(kafkaConfiguration, false, false)
                 .build();
 
         assertThat(configuration, isEquivalent("node.id=2",
@@ -559,7 +576,7 @@ public void testUserConfigurationWithConfigProviders()  {
 
         // Controller
         configuration = new KafkaBrokerConfigurationBuilder(Reconciliation.DUMMY_RECONCILIATION, new NodeRef("my-cluster-kafka-3", 3, "kafka", true, false))
-                .withUserConfiguration(kafkaConfiguration, false)
+                .withUserConfiguration(kafkaConfiguration, false, false)
                 .build();
 
         assertThat(configuration, isEquivalent("node.id=3",
@@ -580,7 +597,7 @@ public void testUserConfigurationWithCCMetricsReporter()  {
         KafkaConfiguration kafkaConfiguration = new KafkaConfiguration(Reconciliation.DUMMY_RECONCILIATION, userConfiguration.entrySet());
 
         String configuration = new KafkaBrokerConfigurationBuilder(Reconciliation.DUMMY_RECONCILIATION, NODE_REF)
-                .withUserConfiguration(kafkaConfiguration, true)
+                .withUserConfiguration(kafkaConfiguration, true, false)
                 .build();
 
         assertThat(configuration, isEquivalent("node.id=2",
@@ -598,6 +615,35 @@ public void testUserConfigurationWithCCMetricsReporter()  {
                 "metric.reporters=com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter"));
     }
 
+    @ParallelTest
+    public void testUserConfigurationWithCCMetricsReporterAndMetricsEnabled()  {
+        Map<String, Object> userConfiguration = new HashMap<>();
+        userConfiguration.put("auto.create.topics.enable", "false");
+        userConfiguration.put("offsets.topic.replication.factor", 3);
+        userConfiguration.put("transaction.state.log.replication.factor", 3);
+        userConfiguration.put("transaction.state.log.min.isr", 2);
+
+        KafkaConfiguration kafkaConfiguration = new KafkaConfiguration(Reconciliation.DUMMY_RECONCILIATION, userConfiguration.entrySet());
+
+        String configuration = new KafkaBrokerConfigurationBuilder(Reconciliation.DUMMY_RECONCILIATION, NODE_REF)
+                .withUserConfiguration(kafkaConfiguration, true, true)
+                .build();
+
+        assertThat(configuration, isEquivalent("node.id=2",
+                "config.providers=strimzienv,strimzifile,strimzidir",
+                "config.providers.strimzienv.class=org.apache.kafka.common.config.provider.EnvVarConfigProvider",
+                "config.providers.strimzienv.param.allowlist.pattern=.*",
+                "config.providers.strimzifile.class=org.apache.kafka.common.config.provider.FileConfigProvider",
+                "config.providers.strimzifile.param.allowed.paths=/opt/kafka",
+                "config.providers.strimzidir.class=org.apache.kafka.common.config.provider.DirectoryConfigProvider",
+                "config.providers.strimzidir.param.allowed.paths=/opt/kafka",
+                "auto.create.topics.enable=false",
+                "offsets.topic.replication.factor=3",
+                "transaction.state.log.replication.factor=3",
+                "transaction.state.log.min.isr=2",
+                "metric.reporters=org.apache.kafka.common.metrics.JmxReporter,com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter"));
+    }
+
     @ParallelTest
     public void testUserConfigurationWithCCMetricsReporterAndOtherMetricReporters()  {
         Map<String, Object> userConfiguration = new HashMap<>();
@@ -606,7 +652,7 @@ public void testUserConfigurationWithCCMetricsReporterAndOtherMetricReporters()
         KafkaConfiguration kafkaConfiguration = new KafkaConfiguration(Reconciliation.DUMMY_RECONCILIATION, userConfiguration.entrySet());
 
         String configuration = new KafkaBrokerConfigurationBuilder(Reconciliation.DUMMY_RECONCILIATION, NODE_REF)
-                .withUserConfiguration(kafkaConfiguration, true)
+                .withUserConfiguration(kafkaConfiguration, true, false)
                 .build();
 
         assertThat(configuration, isEquivalent("node.id=2",
@@ -620,6 +666,28 @@ public void testUserConfigurationWithCCMetricsReporterAndOtherMetricReporters()
                 "metric.reporters=my.domain.CustomMetricReporter,com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter"));
     }
 
+    @ParallelTest
+    public void testUserConfigurationWithCCMetricsReporterAndMetricsEnabledAndOtherMetricReporters()  {
+        Map<String, Object> userConfiguration = new HashMap<>();
+        userConfiguration.put("metric.reporters", "my.domain.CustomMetricReporter");
+
+        KafkaConfiguration kafkaConfiguration = new KafkaConfiguration(Reconciliation.DUMMY_RECONCILIATION, userConfiguration.entrySet());
+
+        String configuration = new KafkaBrokerConfigurationBuilder(Reconciliation.DUMMY_RECONCILIATION, NODE_REF)
+                .withUserConfiguration(kafkaConfiguration, true, true)
+                .build();
+
+        assertThat(configuration, isEquivalent("node.id=2",
+                "config.providers=strimzienv,strimzifile,strimzidir",
+                "config.providers.strimzienv.class=org.apache.kafka.common.config.provider.EnvVarConfigProvider",
+                "config.providers.strimzienv.param.allowlist.pattern=.*",
+                "config.providers.strimzifile.class=org.apache.kafka.common.config.provider.FileConfigProvider",
+                "config.providers.strimzifile.param.allowed.paths=/opt/kafka",
+                "config.providers.strimzidir.class=org.apache.kafka.common.config.provider.DirectoryConfigProvider",
+                "config.providers.strimzidir.param.allowed.paths=/opt/kafka",
+                "metric.reporters=my.domain.CustomMetricReporter,org.apache.kafka.common.metrics.JmxReporter,com.linkedin.kafka.cruisecontrol.metricsreporter.CruiseControlMetricsReporter"));
+    }
+
     @ParallelTest
     public void testEphemeralStorageLogDirs()  {
         Storage storage = new EphemeralStorageBuilder()