fix: loose ends and cleanups (#900)

* group kerberos code together

* clean up bootstrap listener ports

* update metrics.py to actually test for the existence of metrics

* read jmx config from server.yaml instead of broker.yaml

* update docs

* update changelog

Author: razvan

CHANGELOG.md

@@ -16,11 +16,17 @@ All notable changes to this project will be documented in this file.
 - Deprecate support for Kafka `3.7.2` ([#892]).
 - BREAKING: The `<cluster>-<role>-<rolegroup>` rolegroup service was replaced with a `<cluster>-<role>-<rolegroup>-headless`
   and `<cluster>-<role>-<rolegroup>-metrics` rolegroup service ([#897]).
+- Small cleanups and updates ([#900])
+  - remove the metrics port from services that don't need it
+  - use the new `server.yaml` for jmx configuration
+  - update metrics tests
+  - update monitoring doc
 
 [#889]: https://github.com/stackabletech/kafka-operator/pull/889
 [#890]: https://github.com/stackabletech/kafka-operator/pull/890
 [#892]: https://github.com/stackabletech/kafka-operator/pull/892
 [#897]: https://github.com/stackabletech/kafka-operator/pull/897
+[#900]: https://github.com/stackabletech/kafka-operator/pull/900
 
 ## [25.7.0] - 2025-07-23
 

docs/modules/kafka/pages/usage-guide/monitoring.adoc

@@ -1,5 +1,7 @@
 = Monitoring
 :description: The managed Kafka instances are automatically configured to export Prometheus metrics.
 
-The managed Kafka instances are automatically configured to export Prometheus metrics.
+The operator sets up all Kafka server instances with a JMX exporter agent.
+The operator also sets up a dedicated metrics service for each role group.
+The name of this service follows the schema `<stacklet name>-<role name>-<group name>-metrics`.
 See xref:operators:monitoring.adoc[] for more details.

rust/operator-binary/src/config/jvm.rs

@@ -51,7 +51,7 @@ fn construct_jvm_args<ConfigFragment>(
         format!("-Xms{java_heap}"),
         format!("-Djava.security.properties={STACKABLE_CONFIG_DIR}/{JVM_SECURITY_PROPERTIES_FILE}"),
         format!(
-            "-javaagent:/stackable/jmx/jmx_prometheus_javaagent.jar={METRICS_PORT}:/stackable/jmx/broker.yaml"
+            "-javaagent:/stackable/jmx/jmx_prometheus_javaagent.jar={METRICS_PORT}:/stackable/jmx/server.yaml"
         ),
     ];
 
@@ -130,7 +130,7 @@ mod tests {
         assert_eq!(
             non_heap_jvm_args,
             "-Djava.security.properties=/stackable/config/security.properties \
-            -javaagent:/stackable/jmx/jmx_prometheus_javaagent.jar=9606:/stackable/jmx/broker.yaml"
+            -javaagent:/stackable/jmx/jmx_prometheus_javaagent.jar=9606:/stackable/jmx/server.yaml"
         );
         assert_eq!(heap_jvm_args, "-Xmx1638m -Xms1638m");
     }
@@ -177,7 +177,7 @@ mod tests {
         assert_eq!(
             non_heap_jvm_args,
             "-Djava.security.properties=/stackable/config/security.properties \
-            -javaagent:/stackable/jmx/jmx_prometheus_javaagent.jar=9606:/stackable/jmx/broker.yaml \
+            -javaagent:/stackable/jmx/jmx_prometheus_javaagent.jar=9606:/stackable/jmx/server.yaml \
             -Dhttps.proxyHost=proxy.my.corp \
             -Djava.net.preferIPv4Stack=true \
             -Dhttps.proxyPort=1234"

rust/operator-binary/src/crd/security.rs

@@ -484,6 +484,16 @@ impl KafkaTlsSecurity {
                 KafkaListenerName::Bootstrap.listener_ssl_truststore_type(),
                 "PKCS12".to_string(),
             );
+            config.insert("sasl.enabled.mechanisms".to_string(), "GSSAPI".to_string());
+            config.insert(
+                "sasl.kerberos.service.name".to_string(),
+                KafkaRole::Broker.kerberos_service_name().to_string(),
+            );
+            config.insert(
+                "sasl.mechanism.inter.broker.protocol".to_string(),
+                "GSSAPI".to_string(),
+            );
+            tracing::debug!("Kerberos configs added: [{:#?}]", config);
         }
 
         // Internal TLS
@@ -545,20 +555,6 @@ impl KafkaTlsSecurity {
             );
         }
 
-        // Kerberos
-        if self.has_kerberos_enabled() {
-            config.insert("sasl.enabled.mechanisms".to_string(), "GSSAPI".to_string());
-            config.insert(
-                "sasl.kerberos.service.name".to_string(),
-                KafkaRole::Broker.kerberos_service_name().to_string(),
-            );
-            config.insert(
-                "sasl.mechanism.inter.broker.protocol".to_string(),
-                "GSSAPI".to_string(),
-            );
-            tracing::debug!("Kerberos configs added: [{:#?}]", config);
-        }
-
         // common
         config.insert(
             Self::INTER_BROKER_LISTENER_NAME.to_string(),

rust/operator-binary/src/resource/listener.rs

@@ -5,10 +5,7 @@ use stackable_operator::{
 };
 
 use crate::{
-    crd::{
-        METRICS_PORT, METRICS_PORT_NAME, role::broker::BrokerConfig, security::KafkaTlsSecurity,
-        v1alpha1,
-    },
+    crd::{role::broker::BrokerConfig, security::KafkaTlsSecurity, v1alpha1},
     kafka_controller::KAFKA_CONTROLLER_NAME,
     utils::build_recommended_labels,
 };
@@ -53,33 +50,27 @@ pub fn build_broker_rolegroup_bootstrap_listener(
             .build(),
         spec: listener::v1alpha1::ListenerSpec {
             class_name: Some(merged_config.bootstrap_listener_class.clone()),
-            ports: Some(listener_ports(kafka_security)),
+            ports: Some(bootstrap_listener_ports(kafka_security)),
             ..listener::v1alpha1::ListenerSpec::default()
         },
         status: None,
     })
 }
 
-/// We only expose client HTTP / HTTPS and Metrics ports.
-fn listener_ports(kafka_security: &KafkaTlsSecurity) -> Vec<listener::v1alpha1::ListenerPort> {
-    let mut ports = vec![
+fn bootstrap_listener_ports(
+    kafka_security: &KafkaTlsSecurity,
+) -> Vec<listener::v1alpha1::ListenerPort> {
+    vec![if kafka_security.has_kerberos_enabled() {
         listener::v1alpha1::ListenerPort {
-            name: METRICS_PORT_NAME.to_string(),
-            port: METRICS_PORT.into(),
+            name: kafka_security.bootstrap_port_name().to_string(),
+            port: kafka_security.bootstrap_port().into(),
             protocol: Some("TCP".to_string()),
-        },
+        }
+    } else {
         listener::v1alpha1::ListenerPort {
             name: kafka_security.client_port_name().to_string(),
             port: kafka_security.client_port().into(),
             protocol: Some("TCP".to_string()),
-        },
-    ];
-    if kafka_security.has_kerberos_enabled() {
-        ports.push(listener::v1alpha1::ListenerPort {
-            name: kafka_security.bootstrap_port_name().to_string(),
-            port: kafka_security.bootstrap_port().into(),
-            protocol: Some("TCP".to_string()),
-        });
-    }
-    ports
+        }
+    }]
 }

tests/templates/kuttl/smoke-kraft/metrics.py

@@ -3,19 +3,19 @@
 import requests
 
 if __name__ == "__main__":
-    result = 0
-
     LOG_LEVEL = "DEBUG"  # if args.debug else 'INFO'
     logging.basicConfig(
         level=LOG_LEVEL,
         format="%(asctime)s %(levelname)s: %(message)s",
         stream=sys.stdout,
     )
 
-    http_code = requests.get(
-        "http://test-kafka-broker-default-metrics:9606"
-    ).status_code
-    if http_code != 200:
-        result = 1
+    response = requests.get("http://test-kafka-broker-default-metrics:9606/metrics")
+
+    assert response.status_code == 200, (
+        f"Expected HTTP return code 200 from the metrics endpoint but got [{response.status_code}]"
+    )
 
-    sys.exit(result)
+    assert "jmx_scrape_error" in response.text, (
+        "Expected metric [jmx_scrape_error] not found"
+    )

tests/templates/kuttl/smoke/metrics.py

@@ -3,19 +3,19 @@
 import requests
 
 if __name__ == "__main__":
-    result = 0
-
     LOG_LEVEL = "DEBUG"  # if args.debug else 'INFO'
     logging.basicConfig(
         level=LOG_LEVEL,
         format="%(asctime)s %(levelname)s: %(message)s",
         stream=sys.stdout,
     )
 
-    http_code = requests.get(
-        "http://test-kafka-broker-default-metrics:9606/metrics"
-    ).status_code
-    if http_code != 200:
-        result = 1
+    response = requests.get("http://test-kafka-broker-default-metrics:9606/metrics")
+
+    assert response.status_code == 200, (
+        f"Expected HTTP return code 200 from the metrics endpoint but got [{response.status_code}]"
+    )
 
-    sys.exit(result)
+    assert "jmx_scrape_error" in response.text, (
+        "Expected metric [jmx_scrape_error] not found"
+    )