feat: expose publishNotReadyAddresses for broker services (#244)

.github/actions/kind-create/action.yaml

@@ -81,7 +81,7 @@ runs:
     id: cloud-provider-kind
     run: |
       echo "Install cloud-provider-kind"
-      go install sigs.k8s.io/cloud-provider-kind@latest
+      go install sigs.k8s.io/cloud-provider-kind@v0.11.1 # renovate: datasource=go depName=sigs.k8s.io/cloud-provider-kind
       kubectl label node e2e-kind-control-plane node.kubernetes.io/exclude-from-external-load-balancers- 
       ~/go/bin/cloud-provider-kind &
     shell: bash

.github/workflows/e2e-test.yaml

@@ -34,7 +34,7 @@ jobs:
       - name: Setup Helm
         uses: azure/setup-helm@v5
         with:
-          version: '4.2.3'
+          version: '4.2.4' # renovate: datasource=github-releases depName=helm/helm
 
       - name: Clean Go module cache
         run: |

AGENTS.md

@@ -73,7 +73,7 @@ Uses Go workspaces with 5+ modules:
 ### Key Makefile Targets
 
 ```bash
-make test                # Run unit tests with envtest
+make test                # Run all local non-e2e tests across modules (uses envtest)
 make test-e2e            # Run end-to-end tests
 make lint                # Run golangci-lint across all modules
 make check               # Run tests and linters
@@ -88,7 +88,7 @@ make run                 # Run operator locally (outside cluster)
 
 ### Important Make Variables
 - `IMG` - Operator image name (default: `ghcr.io/adobe/koperator:latest`)
-- `ENVTEST_K8S_VERSION` - Kubernetes version for tests (default: 1.31.x)
+- `ENVTEST_K8S_VERSION` - Kubernetes version for envtest assets (default: 1.36.2)
 
 ## Development Workflows
 
@@ -222,9 +222,12 @@ Owner references set:
 Framework: Standard Go testing + testify assertions
 
 ```bash
-# Run all unit tests
+# Run all local non-e2e tests across modules. This is the default test command agents should use.
 make test
 
+# Run tests plus lint before handing off broader changes
+make check
+
 # Run specific package tests
 go test ./pkg/resources/kafka/...
 

Dockerfile

@@ -1,5 +1,5 @@
 # Build the manager binary
-FROM --platform=${BUILDPLATFORM:-linux/amd64} golang:1.26@sha256:2005724102f45917a63e9d092fc0e4ea56ea575048ce147caad5f5f61502c365 AS builder
+FROM --platform=${BUILDPLATFORM:-linux/amd64} golang:1.27@sha256:65b6f280bf050ec5af12716857e8ea8439d694dbba8f31ceeb7630670071f2bb AS builder
 
 ARG BUILDPLATFORM
 ARG TARGETPLATFORM

Makefile

@@ -18,7 +18,7 @@ RELEASE_MSG ?= "koperator release"
 REL_TAG = $(shell ./scripts/increment_version.sh -${RELEASE_TYPE} ${TAG})
 
 # Version constants
-GOLANGCI_VERSION = 2.12.2 # renovate: datasource=github-releases depName=golangci/golangci-lint
+GOLANGCI_VERSION = 2.13.0 # renovate: datasource=github-releases depName=golangci/golangci-lint
 LICENSEI_VERSION = 0.9.0 # renovate: datasource=github-releases depName=goph/licensei
 CONTROLLER_GEN_VERSION = v0.21.0 # renovate: datasource=github-releases depName=kubernetes-sigs/controller-tools
 ENVTEST_K8S_VERSION = 1.36.2 # renovate: datasource=github-releases depName=kubernetes-sigs/controller-tools extractVersion=^envtest-v(?<version>.+)$
@@ -203,6 +203,9 @@ manager: generate fmt vet ## Generate (kubebuilder) and build manager binary.
 run: generate fmt vet
 	go run ./main.go
 
+run-dev: 
+	go run ./main.go --disable-cert-signing-support --disable-webhooks 2>&1 | jq  -Cc -R  'fromjson? // empty' 
+
 # Install CRDs into a cluster by manually creating or replacing the CRD depending on whether is currently existing
 # Apply is not applicable as the last-applied-configuration annotation would exceed the size limit enforced by the api server
 install: manifests ## Install generated CRDs into the configured Kubernetes cluster.

api/assets/kafka/jmx-exporter.yml

@@ -30,8 +30,8 @@ rules:
   # END: Export Kraft metrics
 
   - pattern: 'kafka.server<type=(app-info), id=(\d+)><>(Version): ([-.~+\w\d]+)'
-    name: kafka_server_$1_$3
-    type: COUNTER
+    name: kafka_server_app_info_version
+    type: GAUGE
     labels:
       broker_id: $2
       version: $4

api/assets/kafka/kraft-controller-healthcheck.sh

@@ -15,38 +15,65 @@
 # limitations under the License.
 
 
-# This script returns a successful exit code (0) if the controller is a follower or leader.  For any other state, it returns a failure exit code (1).
-# In addition, if the environment variable KRAFT_HEALTH_CHECK_SKIP is set to "true" (case insensitive), the script will exit successfully without performing any checks.
+# Health check for a KRaft controller, driven by the Prometheus JMX exporter's raft current-state gauge.
+# The jmx-exporter maps the "raft-metrics current-state=<state>" MBean attribute to a single gauge named
+# kafka_server_raft_metrics_current_state_<state> with value 1, so exactly one such gauge exists at a time
+# and its name is the controller's current raft state (leader, follower, candidate, unattached, ...).
+#
+# Behaviour is selected by KRAFT_HEALTH_CHECK_MODE:
+#   readiness        -> fail-closed: succeeds only when the controller is a leader or follower (a
+#                       functioning quorum member). Koperator's rolling upgrade waits on this readiness
+#                       signal so it never restarts the next controller before the previously restarted
+#                       one has rejoined the metadata quorum.
+#   liveness (default) -> fails only when the controller is reachable and reporting a state that is not
+#                       leader/follower; a not-yet-emitted state (startup/catch-up) or an unreachable
+#                       metrics endpoint is treated as healthy (fail-open) so a slow-starting or briefly
+#                       unavailable JMX exporter does not restart an otherwise healthy controller.
+# If KRAFT_HEALTH_CHECK_SKIP is set to "true" (case insensitive) all checks are skipped.
 
 skip_check=$(echo "$KRAFT_HEALTH_CHECK_SKIP" | tr '[:upper:]' '[:lower:]')
+mode=$(echo "${KRAFT_HEALTH_CHECK_MODE:-liveness}" | tr '[:upper:]' '[:lower:]')
 
 if [ "$skip_check" = "true" ]; then
-    echo "KRAFT_HEALTH_CHECK_SKIP is set to TRUE. Exiting health check."
+    echo "KRAFT_HEALTH_CHECK_SKIP is set to TRUE. Skipping health check."
     exit 0
 fi
 
-JMX_ENDPOINT="http://localhost:9020/metrics"
 METRIC_PREFIX="kafka_server_raft_metrics_current_state_"
 
-# Fetch the matching current-state metric with value of 1.0 from the JMX endpoint
-MATCHING_METRIC=$(curl -s "$JMX_ENDPOINT" | grep "^${METRIC_PREFIX}" | awk '$2 == 1.0 {print $1}')
-
-# If it's not empty, it means we found a metric with a value of 1.0.
-if [ -n "$MATCHING_METRIC" ]; then
-    # Determine the state of the controller using the last field name of the metric 
-    # Possible values are leader, candidate, voted, follower, unattached, observer
-    STATE=$(echo "$MATCHING_METRIC" | rev | cut -d'_' -f1 | rev)
-
-    # Check if the extracted state is 'leader' or 'follower'
-    if [ "$STATE" == "leader" ] || [ "$STATE" == "follower" ]; then
-        echo "The controller is in a healthy quorum state."
-        exit 0
-    else
-        # Any other state (e.g., 'candidate', 'unattached', 'observer') is not considered healthy
-        echo "Failure: The controller is in an unexpected state: $STATE. Expecting 'leader' or 'follower'."
-        exit 1
-    fi
-else
-    echo "JMX Exporter endpoint is not avaiable or kafka_server_raft_metrics_current_state_ was not found."
+# Request only the raft current-state gauges via the Prometheus name[] query filter, so the exporter
+# returns a few lines instead of the whole /metrics exposition (which on a combined broker+controller
+# node is large). Only the node's current state is ever present, but we list all known raft states so we
+# can still tell "reporting a non-leader/follower state" apart from "no state emitted yet". If the
+# exporter version ignores name[] it returns the full exposition and the same greps below still work.
+STATES="unattached voted prospective candidate leader follower observer resigned"
+FILTER=""
+for state in ${STATES}; do
+    FILTER="${FILTER}&name[]=${METRIC_PREFIX}${state}"
+done
+URL="http://localhost:9020/metrics?${FILTER#&}"
+
+if ! METRICS=$(curl -s --max-time 4 "${URL}"); then
+    echo "JMX exporter metrics endpoint is not reachable."
+    # Unreachable: not a confirmed quorum member (not ready), but do not restart on a transient blip.
+    [ "${mode}" = "readiness" ] && exit 1
+    exit 0
+fi
+
+# Healthy quorum member: the leader or follower current-state gauge is present.
+if echo "${METRICS}" | grep -Eq "^${METRIC_PREFIX}(leader|follower) 1(\.[0-9]+)?$"; then
+    echo "The controller is in a healthy quorum state (leader or follower)."
     exit 0
 fi
+
+# Reachable and reporting some other raft state (e.g. candidate, unattached, observer).
+if echo "${METRICS}" | grep -q "^${METRIC_PREFIX}"; then
+    STATE=$(echo "${METRICS}" | grep "^${METRIC_PREFIX}" | head -n 1 | sed -E "s/^${METRIC_PREFIX}([a-z]+).*/\1/")
+    echo "Failure: the controller is in an unexpected state: ${STATE}. Expecting 'leader' or 'follower'."
+    exit 1
+fi
+
+# Reachable but no raft current-state gauge yet (startup / not caught up).
+echo "kafka_server_raft_metrics_current_state_ was not found (controller not caught up yet)."
+[ "${mode}" = "readiness" ] && exit 1
+exit 0

api/v1beta1/kafkacluster_types.go

@@ -58,7 +58,7 @@ const (
 	DefaultCruiseControlImage = "adobe/cruise-control:3.0.3-adbe-20260722" // renovate: datasource=docker depName=adobe/cruise-control
 
 	// DefaultKafkaImage is the default Kafka image used when users don't specify it in KafkaClusterSpec.ClusterImage
-	DefaultKafkaImage = "ghcr.io/adobe/koperator/kafka:2.13-3.9.1" // renovate: datasource=docker depName=ghcr.io/adobe/koperator/kafka
+	DefaultKafkaImage = "ghcr.io/adobe/koperator/kafka:2.13-3.9.2-jdk21.0.11" // renovate: datasource=docker depName=ghcr.io/adobe/koperator/kafka
 
 	// DefaultMonitorImage is the default JMX monitor image used when users don't specify it in MonitoringConfig.JmxImage
 	DefaultMonitorImage = "ghcr.io/adobe/koperator/jmx-javaagent:1.5.0" // renovate: datasource=docker depName=ghcr.io/adobe/koperator/jmx-javaagent
@@ -131,7 +131,7 @@ const (
 	defaultKafkaClusterK8sClusterDomain  = "cluster.local"
 
 	// KafkaBroker.spec.container["kafka"].image
-	defaultKafkaImage = "ghcr.io/adobe/koperator/kafka:2.13-3.9.1" // renovate: datasource=docker depName=ghcr.io/adobe/koperator/kafka
+	defaultKafkaImage = "ghcr.io/adobe/koperator/kafka:2.13-3.9.2-jdk21.0.11" // renovate: datasource=docker depName=ghcr.io/adobe/koperator/kafka
 
 	/* Monitor Config */
 
@@ -148,6 +148,16 @@ type KafkaClusterSpec struct {
 	// +optional
 	KRaftMode              bool `json:"kRaft"`
 	HeadlessServiceEnabled bool `json:"headlessServiceEnabled"`
+	// PublishNotReadyAddresses controls whether the non-headless broker Services
+	// (kafka-{id} and kafka-all-broker) include the pod IP in their endpoints even
+	// when the pod fails its readiness probe.
+	// Enable this when using a custom readiness sidecar that checks Under-Replicated
+	// Partitions (URP) combined with a PodDisruptionBudget, so that brokers continue
+	// serving existing partition traffic while the PDB holds eviction.
+	// Headless services always publish not-ready addresses regardless of this field.
+	// +kubebuilder:default=false
+	// +optional
+	PublishNotReadyAddresses bool `json:"publishNotReadyAddresses,omitempty"`
 	// localDebugEnabled is used to decide whether to create a separate loadbalancer services for the
 	// Kafka and Cruise Control Pods. These services will expose the internal listener ports of the Kafka
 	// cluster with LoadBalancer type, which can be used for running Koperator on a local machine against

charts/kafka-operator/crds/kafkaclusters.yaml

@@ -22965,6 +22965,17 @@ spec:
                 type: boolean
               propagateLabels:
                 type: boolean
+              publishNotReadyAddresses:
+                default: false
+                description: |-
+                  PublishNotReadyAddresses controls whether the non-headless broker Services
+                  (kafka-{id} and kafka-all-broker) include the pod IP in their endpoints even
+                  when the pod fails its readiness probe.
+                  Enable this when using a custom readiness sidecar that checks Under-Replicated
+                  Partitions (URP) combined with a PodDisruptionBudget, so that brokers continue
+                  serving existing partition traffic while the PDB holds eviction.
+                  Headless services always publish not-ready addresses regardless of this field.
+                type: boolean
               rackAwareness:
                 description: RackAwareness defines the required fields to enable kafka's
                   rack aware feature

charts/kafka-operator/templates/alertmanager-peerauthentication.yaml

@@ -17,7 +17,7 @@ spec:
   selector:
     matchLabels:
       control-plane: controller-manager
-      component: alertmanager
+      app.kubernetes.io/component: operator
   portLevelMtls:
     {{ .Values.alertManager.port | quote }}:
       mode: PERMISSIVE

charts/kafka-operator/templates/operator-deployment-with-webhook.yaml

@@ -133,8 +133,6 @@ spec:
         app.kubernetes.io/name: {{ include "kafka-operator.name" . }}
         app.kubernetes.io/instance: {{ .Release.Name }}
         app.kubernetes.io/component: operator
-        app: prometheus
-        component: alertmanager
     spec:
     {{- with .Values.imagePullSecrets }}
       imagePullSecrets:

charts/kafka-operator/templates/podmonitor.yaml

@@ -0,0 +1,31 @@
+{{- if .Values.prometheusMetrics.podMonitor.enabled }}
+kind: PodMonitor
+apiVersion: monitoring.coreos.com/v1
+metadata:
+  name: {{ include "kafka-operator.fullname" . }}
+  namespace: {{ .Release.Namespace | quote }}
+  labels:
+    helm.sh/chart: {{ include "kafka-operator.chart" . }}
+    app.kubernetes.io/name: {{ include "kafka-operator.name" . }}
+    app.kubernetes.io/instance: {{ .Release.Name }}
+    app.kubernetes.io/managed-by: {{ .Release.Service }}
+    app.kubernetes.io/version: {{ .Chart.AppVersion }}
+    app.kubernetes.io/component: operator
+  {{- with .Values.operator.annotations }}
+  annotations:
+    {{- toYaml . | nindent 4 }}
+  {{- end }}
+spec:
+  namespaceSelector:
+    matchNames:
+    - {{ .Release.Namespace }}
+  selector:
+    matchLabels:
+      app.kubernetes.io/name: {{ include "kafka-operator.name" . }}
+      app.kubernetes.io/instance: {{ .Release.Name }}
+      app.kubernetes.io/component: operator
+  endpoints:
+  - interval: {{ .Values.prometheusMetrics.podMonitor.interval }}
+    port: metrics
+    path: /metrics
+{{- end }}

charts/kafka-operator/values.yaml

@@ -111,6 +111,10 @@ prometheusMetrics:
       create: true
       # -- ServiceAccount used by prometheus auth proxy
       name: kafka-operator-authproxy
+  podMonitor:
+    # -- If true, create a PodMonitor for Prometheus metrics
+    enabled: false
+    interval: 30s
 
 # -- Health probes configuration
 healthProbes: {}

config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml

@@ -22965,6 +22965,17 @@ spec:
                 type: boolean
               propagateLabels:
                 type: boolean
+              publishNotReadyAddresses:
+                default: false
+                description: |-
+                  PublishNotReadyAddresses controls whether the non-headless broker Services
+                  (kafka-{id} and kafka-all-broker) include the pod IP in their endpoints even
+                  when the pod fails its readiness probe.
+                  Enable this when using a custom readiness sidecar that checks Under-Replicated
+                  Partitions (URP) combined with a PodDisruptionBudget, so that brokers continue
+                  serving existing partition traffic while the PDB holds eviction.
+                  Headless services always publish not-ready addresses regardless of this field.
+                type: boolean
               rackAwareness:
                 description: RackAwareness defines the required fields to enable kafka's
                   rack aware feature

config/base/manager/manager.yaml

@@ -20,7 +20,7 @@ spec:
         - /manager
         args:
         - --enable-leader-election
-        image: ghcr.io/adobe/koperator:latest
+        image: ghcr.io/adobe/koperator:latest # renovate: datasource=docker depName=ghcr.io/adobe/koperator
         name: manager
         livenessProbe:
           initialDelaySeconds: 15

config/overlays/auth-proxy-enabled/manager_auth_proxy_patch.yaml

@@ -10,7 +10,7 @@ spec:
     spec:
       containers:
       - name: kube-rbac-proxy
-        image: quay.io/brancz/kube-rbac-proxy:v0.20.0
+        image: quay.io/brancz/kube-rbac-proxy:v0.22.1 # renovate: datasource=docker depName=quay.io/brancz/kube-rbac-proxy
         args:
         - "--secure-listen-address=0.0.0.0:8443"
         - "--upstream=http://127.0.0.1:8080/"

config/overlays/certmanager-with-auth-proxy/auth-proxy/manager_auth_proxy_patch.yaml

@@ -10,7 +10,7 @@ spec:
     spec:
       containers:
       - name: kube-rbac-proxy
-        image: quay.io/brancz/kube-rbac-proxy:v0.20.0
+        image: quay.io/brancz/kube-rbac-proxy:v0.22.1 # renovate: datasource=docker depName=quay.io/brancz/kube-rbac-proxy
         args:
         - "--secure-listen-address=0.0.0.0:8443"
         - "--upstream=http://127.0.0.1:8080/"

config/samples/banzaicloud_v1beta1_kafkacluster.yaml

@@ -252,7 +252,7 @@ spec:
         #volumeMounts:
         #  - mountPath: "/opt/kafka/libs/extra-jars"
         #    name: "jars"
-    - image: "ghcr.io/adobe/koperator/kafka:2.13-3.9.1"
+    - image: "ghcr.io/adobe/koperator/kafka:2.13-3.9.2-jdk21.0.11" # renovate: datasource=docker depName=ghcr.io/adobe/koperator/kafka
       id: 1
       config: |
         auto.create.topics.enable=false
@@ -274,7 +274,7 @@ spec:
       readOnlyConfig: |
         auto.create.topics.enable=false
       brokerConfig:
-        image: "ghcr.io/adobe/koperator/kafka:2.13-3.9.1"
+        image: "ghcr.io/adobe/koperator/kafka:2.13-3.9.2-jdk21.0.11" # renovate: datasource=docker depName=ghcr.io/adobe/koperator/kafka
         storageConfigs:
           - mountPath: "/kafka-logs"
             pvcSpec:

config/samples/kafkacluster-prometheus.yaml

@@ -85,7 +85,7 @@ spec:
         storageClass: 'gp2'
         mountPath: '/kafkalog'
         diskSize: '10G'
-        image: 'ghcr.io/adobe/koperator/kafka:2.13-3.9.1'
+        image: 'ghcr.io/adobe/koperator/kafka:2.13-3.9.2-jdk21.0.11' # renovate: datasource=docker depName=ghcr.io/adobe/koperator/kafka
         # annotations to be applied onto the broker pod
         # brokerAnnotations: '{ "sidecar.istio.io/logLevel": "trace", "sidecar.istio.io/proxyCPULimit": "50m" }'
         command: 'upScale'
@@ -101,7 +101,7 @@ spec:
         storageClass: 'gp2'
         mountPath: '/kafkalog'
         diskSize: '10G'
-        image: 'ghcr.io/adobe/koperator/kafka:2.13-3.9.1'
+        image: 'ghcr.io/adobe/koperator/kafka:2.13-3.9.2-jdk21.0.11' # renovate: datasource=docker depName=ghcr.io/adobe/koperator/kafka
         # annotations to be applied onto the broker pod
         # brokerAnnotations: '{ "sidecar.istio.io/logLevel": "trace", "sidecar.istio.io/proxyCPULimit": "50m" }'
         command: 'upScale'
@@ -117,7 +117,7 @@ spec:
         storageClass: 'gp2'
         mountPath: '/kafkalog'
         diskSize: '10G'
-        image: 'ghcr.io/adobe/koperator/kafka:2.13-3.9.1'
+        image: 'ghcr.io/adobe/koperator/kafka:2.13-3.9.2-jdk21.0.11' # renovate: datasource=docker depName=ghcr.io/adobe/koperator/kafka
         # annotations to be applied onto the broker pod
         # brokerAnnotations: '{ "sidecar.istio.io/logLevel": "trace", "sidecar.istio.io/proxyCPULimit": "50m" }'
         command: 'upScale'

config/samples/kafkacluster_with_external_ssl.yaml

@@ -10,7 +10,7 @@ spec:
   zkAddresses:
     - "zookeeper-server-client.zookeeper:2181"
   oneBrokerPerNode: false
-  clusterImage: "ghcr.io/adobe/koperator/kafka:2.13-3.9.1"
+  clusterImage: "ghcr.io/adobe/koperator/kafka:2.13-3.9.2-jdk21.0.11" # renovate: datasource=docker depName=ghcr.io/adobe/koperator/kafka
   readOnlyConfig: |
     auto.create.topics.enable=false
     cruise.control.metrics.topic.auto.create=true

config/samples/kafkacluster_with_external_ssl_customcert.yaml

@@ -10,7 +10,7 @@ spec:
   zkAddresses:
     - "zookeeper-server-client.zookeeper:2181"
   oneBrokerPerNode: false
-  clusterImage: "ghcr.io/adobe/koperator/kafka:2.13-3.9.1"
+  clusterImage: "ghcr.io/adobe/koperator/kafka:2.13-3.9.2-jdk21.0.11" # renovate: datasource=docker depName=ghcr.io/adobe/koperator/kafka
   readOnlyConfig: |
     auto.create.topics.enable=false
     cruise.control.metrics.topic.auto.create=true