Strimzi Metrics Reporter integration with KafkaConnect and MirrorMaker2

api/src/main/java/io/strimzi/api/kafka/model/connect/AbstractKafkaConnectSpec.java

@@ -23,7 +23,6 @@
 import io.strimzi.api.kafka.model.common.jmx.KafkaJmxOptions;
 import io.strimzi.api.kafka.model.common.metrics.MetricsConfig;
 import io.strimzi.api.kafka.model.common.tracing.Tracing;
-import io.strimzi.crdgenerator.annotations.CelValidation;
 import io.strimzi.crdgenerator.annotations.Description;
 import io.strimzi.crdgenerator.annotations.KubeLink;
 import io.sundr.builder.annotations.Buildable;
@@ -156,18 +155,8 @@ public void setJmxOptions(KafkaJmxOptions jmxOptions) {
         this.jmxOptions = jmxOptions;
     }
 
-    @Description("Metrics configuration. Only `jmxPrometheusExporter` can be configured, as this component does not yet support `strimziMetricsReporter`.")
+    @Description("Metrics configuration.")
     @JsonInclude(JsonInclude.Include.NON_EMPTY)
-    @CelValidation(rules = {
-        @CelValidation.CelValidationRule(
-                rule = "self.type != 'jmxPrometheusExporter' || has(self.valueFrom)",
-                message = "valueFrom property is required"
-            ),
-        @CelValidation.CelValidationRule(
-                rule = "self.type != 'strimziMetricsReporter'",
-                message = "value type not supported"
-            )
-    })
     @Override
     public MetricsConfig getMetricsConfig() {
         return metricsConfig;

api/src/main/java/io/strimzi/api/kafka/model/connect/KafkaConnectSpec.java

@@ -33,7 +33,7 @@
 @EqualsAndHashCode(callSuper = true, doNotUseGetters = true)
 @ToString(callSuper = true)
 public class KafkaConnectSpec extends AbstractKafkaConnectSpec {
-    public static final String FORBIDDEN_PREFIXES = "ssl., sasl., security., listeners, plugin.path, rest., bootstrap.servers, consumer.interceptor.classes, producer.interceptor.classes";
+    public static final String FORBIDDEN_PREFIXES = "ssl., sasl., security., listeners, plugin.path, rest., bootstrap.servers, consumer.interceptor.classes, producer.interceptor.classes, prometheus.metrics.reporter.";
     public static final String FORBIDDEN_PREFIX_EXCEPTIONS = "ssl.endpoint.identification.algorithm, ssl.cipher.suites, ssl.protocol, ssl.enabled.protocols";
 
     private Map<String, Object> config = new HashMap<>(0);

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

@@ -31,6 +31,7 @@
 import io.strimzi.kafka.oauth.server.plain.ServerPlainConfig;
 import io.strimzi.operator.cluster.model.cruisecontrol.CruiseControlMetricsReporter;
 import io.strimzi.operator.cluster.model.metrics.MetricsModel;
+import io.strimzi.operator.cluster.model.metrics.StrimziMetricsReporterConfig;
 import io.strimzi.operator.cluster.model.metrics.StrimziMetricsReporterModel;
 import io.strimzi.operator.common.InvalidConfigurationException;
 import io.strimzi.operator.common.Reconciliation;
@@ -45,7 +46,6 @@
 import java.util.Comparator;
 import java.util.HashMap;
 import java.util.LinkedHashMap;
-import java.util.LinkedHashSet;
 import java.util.List;
 import java.util.Locale;
 import java.util.Map;
@@ -66,8 +66,6 @@ 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;
@@ -145,16 +143,16 @@ public KafkaBrokerConfigurationBuilder withCruiseControl(String clusterName, Cru
     /**
      * Configures the Strimzi Metrics Reporter. It is set only if user enables Strimzi Metrics Reporter.
      *
-     * @param model     Strimzi Metrics Reporter configuration
+     * @param model Strimzi Metrics Reporter configuration
      *
      * @return Returns the builder instance
      */
-    public KafkaBrokerConfigurationBuilder withStrimziMetricsReporter(MetricsModel model)   {
+    public KafkaBrokerConfigurationBuilder withStrimziMetricsReporter(MetricsModel model) {
         if (model instanceof StrimziMetricsReporterModel reporterModel) {
             printSectionHeader("Strimzi Metrics Reporter configuration");
-            writer.println("prometheus.metrics.reporter.listener.enable=true");
-            writer.println("prometheus.metrics.reporter.listener=http://:" + StrimziMetricsReporterModel.METRICS_PORT);
-            writer.println("prometheus.metrics.reporter.allowlist=" + reporterModel.getAllowList());
+            writer.println(StrimziMetricsReporterConfig.LISTENER_ENABLE + "=true");
+            writer.println(StrimziMetricsReporterConfig.LISTENER + "=http://:" + MetricsModel.METRICS_PORT);
+            writer.println(StrimziMetricsReporterConfig.ALLOW_LIST + "=" + reporterModel.getAllowList());
             writer.println();
         }
         return this;
@@ -829,7 +827,7 @@ private String getConfigProviderAliases(KafkaConfiguration userConfig) {
     /**
      * Get the user provided Kafka configuration provider aliases, throwing an InvalidConfigurationException if any are found that would overwrite the Strimzi defined configuration providers
      * 
-     * @param strimziConfigProviders    The Strimzi defined configuration providers
+     * @param strimziAliases            The Strimzi defined configuration providers
      * @param userConfig                The user configuration to extract the possible user-provided config provider configuration from
      * @return                          The user defined Kafka configuration provider aliases or empty string
      */
@@ -854,10 +852,10 @@ private String getUserConfigProviderAliases(Collection<String> strimziAliases, K
     /**
      * Adds the configurations passed by the user in the Kafka CR, injecting Strimzi configurations when needed.
      *
-     * @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 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 injectKafkaJmxReporter         Flag to indicate if metrics are enabled. If they are we inject the JmxReporter into the configuration
      * @param injectStrimziMetricsReporter   Inject the Strimzi Metrics Reporter into the configuration
-     * @param injectKafkaJmxReporter          Flag to indicate if metrics are enabled. If they are we inject the JmxReporter into the configuration
      *
      * @return Returns the builder instance
      */
@@ -872,13 +870,9 @@ public KafkaBrokerConfigurationBuilder withUserConfiguration(KafkaConfiguration
 
         configProviders(userConfig);
 
-        // Adds the Kafka metric.reporters to the user configuration.
-        maybeAddMetricReporters(userConfig, injectCcMetricsReporter, injectKafkaJmxReporter, injectStrimziMetricsReporter);
-
-        // Adds the Yammer kafka.metrics.reporters to the user configuration.
-        maybeAddYammerMetricsReporters(userConfig, injectStrimziMetricsReporter);
+        printMetricReporters(userConfig, injectCcMetricsReporter, injectKafkaJmxReporter, injectStrimziMetricsReporter);
+        printYammerReporters(userConfig, injectStrimziMetricsReporter);
 
-        // print user config with Strimzi injections
         if (!userConfig.getConfiguration().isEmpty()) {
             printSectionHeader("User provided configuration");
             writer.println(userConfig.getConfiguration());
@@ -888,35 +882,70 @@ public KafkaBrokerConfigurationBuilder withUserConfiguration(KafkaConfiguration
         return this;
     }
 
-    /**
-     * Adds the Kafka metric.reporters to the user configuration.
-     *
-     * @param userConfig The user configuration to which the metric reporters will be added.
-     * @param injectCcMetricsReporter Flag indicating whether to inject the Cruise Control Metrics Reporter.
-     * @param injectKafkaJmxReporter  Inject the JMX Reporter into the configuration
-     * @param injectStrimziMetricsReporter Flag indicating whether to inject the Strimzi Metrics Reporter.
-     */
-    private void maybeAddMetricReporters(KafkaConfiguration userConfig, boolean injectCcMetricsReporter, boolean injectKafkaJmxReporter, boolean injectStrimziMetricsReporter) {
-        if (injectCcMetricsReporter) {
-            createOrAddListConfig(userConfig, "metric.reporters", CruiseControlMetricsReporter.CRUISE_CONTROL_METRIC_REPORTER);
-        }
-        if (injectKafkaJmxReporter) {
-            createOrAddListConfig(userConfig, "metric.reporters", "org.apache.kafka.common.metrics.JmxReporter");
-        }
-        if (injectStrimziMetricsReporter) {
-            createOrAddListConfig(userConfig, "metric.reporters", "io.strimzi.kafka.metrics.KafkaPrometheusMetricsReporter");
+    private void printMetricReporters(KafkaConfiguration userConfig,
+                                      boolean injectCcMetricsReporter,
+                                      boolean injectKafkaJmxReporter,
+                                      boolean injectStrimziMetricsReporter) {
+        // Build a list of reporters to inject based on flags
+        List<String> reportersToInject = new ArrayList<>();
+        if (injectCcMetricsReporter) reportersToInject.add(CruiseControlMetricsReporter.CRUISE_CONTROL_METRIC_REPORTER);
+        // Since Kafka 4 the JmxReporter is explicitly added when JmxPrometheusExporter is enabled
+        if (injectKafkaJmxReporter) reportersToInject.add("org.apache.kafka.common.metrics.JmxReporter");
+        if (injectStrimziMetricsReporter) reportersToInject.add(StrimziMetricsReporterConfig.KAFKA_CLASS);
+
+        if (!reportersToInject.isEmpty()) {
+            String configKey = "metric.reporters";
+
+            if (userConfig != null && !userConfig.getConfiguration().isEmpty() &&
+                    userConfig.getConfigOption(configKey) != null) {
+                // handle user configuration if present and avoids duplicates
+                String configValue = userConfig.getConfigOption(configKey);
+
+                reportersToInject = reportersToInject.stream()
+                        .filter(r -> !userConfig.getConfigOption(configKey).contains(r))
+                        .toList();
+
+                if (!reportersToInject.isEmpty()) {
+                    userConfig.removeConfigOption(configKey);
+
+                    printSectionHeader(configKey + " configuration");
+                    writer.println("# " + configKey + " configured by the user and by Strimzi");
+                    writer.println(configKey + "=" + configValue + "," + String.join(",", reportersToInject));
+                    writer.println();
+                }
+            } else {
+                printSectionHeader(configKey + " configuration");
+                writer.println("# " + configKey + " configured by Strimzi");
+                writer.println(configKey + "=" + String.join(",", reportersToInject));
+                writer.println();
+            }
         }
     }
 
-    /**
-     * Adds the Yammer kafka.metrics.reporters to the user configuration if the Strimzi Metrics Reporter is enabled.
-     *
-     * @param userConfig The user configuration to which the Yammer metrics reporter will be added.
-     * @param injectStrimziMetricsReporter Flag indicating whether to inject the Strimzi Metrics Reporter.
-     */
-    private void maybeAddYammerMetricsReporters(KafkaConfiguration userConfig, boolean injectStrimziMetricsReporter) {
+    private void printYammerReporters(KafkaConfiguration userConfig,
+                                      boolean injectStrimziMetricsReporter) {
         if (injectStrimziMetricsReporter) {
-            createOrAddListConfig(userConfig, "kafka.metrics.reporters", "io.strimzi.kafka.metrics.YammerPrometheusMetricsReporter");
+            String configKey = "kafka.metrics.reporters";
+
+            if (userConfig != null
+                    && !userConfig.getConfiguration().isEmpty()
+                    && userConfig.getConfigOption(configKey) != null) {
+                // handle user configuration if present and avoids duplicates
+                String configValue = userConfig.getConfigOption(configKey);
+
+                if (!configValue.contains(StrimziMetricsReporterConfig.YAMMER_CLASS)) {
+                    userConfig.removeConfigOption(configKey);
+                    printSectionHeader(configKey + " configuration");
+                    writer.println("# " + configKey + " configured by the user and by Strimzi");
+                    writer.println(configKey + "=" + configValue + "," + StrimziMetricsReporterConfig.YAMMER_CLASS);
+                    writer.println();
+                }
+            } else {
+                printSectionHeader(configKey + " configuration");
+                writer.println("# " + configKey + " configured by Strimzi");
+                writer.println(configKey + "=" + StrimziMetricsReporterConfig.YAMMER_CLASS);
+                writer.println();
+            }
         }
     }
 
@@ -1061,45 +1090,6 @@ private void configureQuotasPluginStrimzi(String clusterName, QuotasPluginStrimz
         writer.println(String.format("client.quota.callback.static.excluded.principal.name.list=%s", String.join(";", excludedPrincipals)));
     }
 
-    /**
-     * Append list configuration values or create a new list configuration if missing.
-     * A list configuration can contain a comma separated list of values.
-     * Duplicated values are removed.
-     *
-     * @param kafkaConfig Kafka configuration.
-     * @param key List configuration key.
-     * @param values List configuration values.
-     */
-    static void createOrAddListConfig(AbstractConfiguration kafkaConfig, String key, String values) {
-        if (kafkaConfig == null) {
-            throw new IllegalArgumentException("Configuration is required");
-        }
-        if (key == null || key.isBlank()) {
-            throw new IllegalArgumentException("Configuration key is required");
-        }
-        if (values == null || values.isBlank()) {
-            throw new IllegalArgumentException("Configuration values are required");
-        }
-
-        String existingConfig = kafkaConfig.getConfigOption(key);
-        // using an ordered set to preserve ordering of the existing kafkaConfig as values could potentially be user-provided.
-        Set<String> existingSet = existingConfig == null ? new LinkedHashSet<>() :
-                Arrays.stream(existingConfig.split(","))
-                        .map(String::trim)
-                        .filter(s -> !s.isEmpty())
-                        .collect(Collectors.toCollection(LinkedHashSet::new));
-        Set<String> newValues = Arrays.stream(values.split(","))
-                .map(String::trim)
-                .filter(s -> !s.isEmpty())
-                .collect(Collectors.toCollection(LinkedHashSet::new));
-        // add only new values
-        boolean updated = existingSet.addAll(newValues);
-        if (updated) {
-            String updatedConfig = String.join(",", existingSet);
-            kafkaConfig.setConfigOption(key, updatedConfig);
-        }
-    }
-
     /**
      * Internal method which prints the section header into the configuration file. This makes it more human-readable
      * when looking for issues in running pods etc.

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

@@ -1838,9 +1838,12 @@ private String generatePerBrokerConfiguration(NodeRef node, KafkaPool pool, Map<
                 .withTieredStorage(cluster, tieredStorage)
                 .withQuotas(cluster, quotas)
                 .withStrimziMetricsReporter(metrics)
-                .withUserConfiguration(configuration, node.broker() && ccMetricsReporter != null,
-                         metrics instanceof JmxPrometheusExporterModel, metrics instanceof StrimziMetricsReporterModel)
-                .build().trim();
+                .withUserConfiguration(
+                        configuration,
+                        node.broker() && ccMetricsReporter != null,
+                        metrics instanceof JmxPrometheusExporterModel,
+                        metrics instanceof StrimziMetricsReporterModel
+                ).build().trim();
     }
 
     /**