diff --git a/.github/actions/kind-create/action.yaml b/.github/actions/kind-create/action.yaml index 12e3034ef..d3dfeb8c5 100644 --- a/.github/actions/kind-create/action.yaml +++ b/.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 diff --git a/.github/workflows/e2e-test.yaml b/.github/workflows/e2e-test.yaml index 3d8f656a0..0ecfe02b4 100644 --- a/.github/workflows/e2e-test.yaml +++ b/.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: | diff --git a/AGENTS.md b/AGENTS.md index a57c41cee..865a47e10 100644 --- a/AGENTS.md +++ b/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/... diff --git a/Dockerfile b/Dockerfile index 4c47dfce9..365318f78 100644 --- a/Dockerfile +++ b/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 diff --git a/Makefile b/Makefile index 3d9ba57ed..4263c3459 100644 --- a/Makefile +++ b/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(?.+)$ @@ -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. diff --git a/api/assets/kafka/jmx-exporter.yml b/api/assets/kafka/jmx-exporter.yml index a3857da29..48d953ef9 100644 --- a/api/assets/kafka/jmx-exporter.yml +++ b/api/assets/kafka/jmx-exporter.yml @@ -30,8 +30,8 @@ rules: # END: Export Kraft metrics - pattern: 'kafka.server<>(Version): ([-.~+\w\d]+)' - name: kafka_server_$1_$3 - type: COUNTER + name: kafka_server_app_info_version + type: GAUGE labels: broker_id: $2 version: $4 diff --git a/api/assets/kafka/kraft-controller-healthcheck.sh b/api/assets/kafka/kraft-controller-healthcheck.sh index f65e88148..5f426a53a 100755 --- a/api/assets/kafka/kraft-controller-healthcheck.sh +++ b/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=" MBean attribute to a single gauge named +# kafka_server_raft_metrics_current_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 diff --git a/api/v1beta1/kafkacluster_types.go b/api/v1beta1/kafkacluster_types.go index 4bc0fe516..6e47f5910 100644 --- a/api/v1beta1/kafkacluster_types.go +++ b/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 diff --git a/charts/kafka-operator/crds/kafkaclusters.yaml b/charts/kafka-operator/crds/kafkaclusters.yaml index a7266064a..b92b352f0 100644 --- a/charts/kafka-operator/crds/kafkaclusters.yaml +++ b/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 diff --git a/charts/kafka-operator/templates/alertmanager-peerauthentication.yaml b/charts/kafka-operator/templates/alertmanager-peerauthentication.yaml index 79f8aa3e2..d1ca50bf4 100644 --- a/charts/kafka-operator/templates/alertmanager-peerauthentication.yaml +++ b/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 diff --git a/charts/kafka-operator/templates/operator-deployment-with-webhook.yaml b/charts/kafka-operator/templates/operator-deployment-with-webhook.yaml index 9d82ddd3e..15dde16ba 100644 --- a/charts/kafka-operator/templates/operator-deployment-with-webhook.yaml +++ b/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: diff --git a/charts/kafka-operator/templates/podmonitor.yaml b/charts/kafka-operator/templates/podmonitor.yaml new file mode 100644 index 000000000..06b8b87a7 --- /dev/null +++ b/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 }} diff --git a/charts/kafka-operator/values.yaml b/charts/kafka-operator/values.yaml index 5d6dfca02..638cfe3a2 100644 --- a/charts/kafka-operator/values.yaml +++ b/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: {} diff --git a/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml b/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml index a7266064a..b92b352f0 100644 --- a/config/base/crds/kafka.banzaicloud.io_kafkaclusters.yaml +++ b/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 diff --git a/config/base/manager/manager.yaml b/config/base/manager/manager.yaml index 5e1702cc9..676e63214 100644 --- a/config/base/manager/manager.yaml +++ b/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 diff --git a/config/overlays/auth-proxy-enabled/manager_auth_proxy_patch.yaml b/config/overlays/auth-proxy-enabled/manager_auth_proxy_patch.yaml index 7c697059b..a58c4d549 100644 --- a/config/overlays/auth-proxy-enabled/manager_auth_proxy_patch.yaml +++ b/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/" diff --git a/config/overlays/certmanager-with-auth-proxy/auth-proxy/manager_auth_proxy_patch.yaml b/config/overlays/certmanager-with-auth-proxy/auth-proxy/manager_auth_proxy_patch.yaml index 978e56bde..c6ccee07b 100644 --- a/config/overlays/certmanager-with-auth-proxy/auth-proxy/manager_auth_proxy_patch.yaml +++ b/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/" diff --git a/config/samples/banzaicloud_v1beta1_kafkacluster.yaml b/config/samples/banzaicloud_v1beta1_kafkacluster.yaml index 1550ad04c..863deb8f8 100644 --- a/config/samples/banzaicloud_v1beta1_kafkacluster.yaml +++ b/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: diff --git a/config/samples/kafkacluster-prometheus.yaml b/config/samples/kafkacluster-prometheus.yaml index 089c84c63..d5715129f 100644 --- a/config/samples/kafkacluster-prometheus.yaml +++ b/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' diff --git a/config/samples/kafkacluster_with_external_ssl.yaml b/config/samples/kafkacluster_with_external_ssl.yaml index c650e28ba..ae4d9f06e 100644 --- a/config/samples/kafkacluster_with_external_ssl.yaml +++ b/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 diff --git a/config/samples/kafkacluster_with_external_ssl_customcert.yaml b/config/samples/kafkacluster_with_external_ssl_customcert.yaml index c8562f942..2e6a0a0bd 100644 --- a/config/samples/kafkacluster_with_external_ssl_customcert.yaml +++ b/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 diff --git a/config/samples/kafkacluster_with_ssl_groups.yaml b/config/samples/kafkacluster_with_ssl_groups.yaml index a5e592555..21398d70d 100644 --- a/config/samples/kafkacluster_with_ssl_groups.yaml +++ b/config/samples/kafkacluster_with_ssl_groups.yaml @@ -15,7 +15,7 @@ spec: - "failure-domain.beta.kubernetes.io/region" - "failure-domain.beta.kubernetes.io/zone" 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 diff --git a/config/samples/kafkacluster_with_ssl_groups_customcert.yaml b/config/samples/kafkacluster_with_ssl_groups_customcert.yaml index 276ebbe1d..e832864e7 100644 --- a/config/samples/kafkacluster_with_ssl_groups_customcert.yaml +++ b/config/samples/kafkacluster_with_ssl_groups_customcert.yaml @@ -15,7 +15,7 @@ spec: - "failure-domain.beta.kubernetes.io/region" - "failure-domain.beta.kubernetes.io/zone" 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 diff --git a/config/samples/kafkacluster_with_ssl_hybrid_customcert.yaml b/config/samples/kafkacluster_with_ssl_hybrid_customcert.yaml index e36ad58bc..fe206a1fe 100644 --- a/config/samples/kafkacluster_with_ssl_hybrid_customcert.yaml +++ b/config/samples/kafkacluster_with_ssl_hybrid_customcert.yaml @@ -15,7 +15,7 @@ spec: - "failure-domain.beta.kubernetes.io/region" - "failure-domain.beta.kubernetes.io/zone" 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 diff --git a/config/samples/kafkacluster_without_ssl.yaml b/config/samples/kafkacluster_without_ssl.yaml index 00e4639d1..1f2dc3fde 100644 --- a/config/samples/kafkacluster_without_ssl.yaml +++ b/config/samples/kafkacluster_without_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 diff --git a/config/samples/kafkacluster_without_ssl_groups.yaml b/config/samples/kafkacluster_without_ssl_groups.yaml index 991421aaa..227c36f18 100644 --- a/config/samples/kafkacluster_without_ssl_groups.yaml +++ b/config/samples/kafkacluster_without_ssl_groups.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 diff --git a/config/samples/kraft/simplekafkacluster_kraft.yaml b/config/samples/kraft/simplekafkacluster_kraft.yaml index 1e993bb49..c69a1ff4c 100644 --- a/config/samples/kraft/simplekafkacluster_kraft.yaml +++ b/config/samples/kraft/simplekafkacluster_kraft.yaml @@ -7,11 +7,14 @@ metadata: spec: kRaft: true monitoringConfig: - jmxImage: "ghcr.io/adobe/koperator/jmx-javaagent:1.4.0" + jmxImage: "ghcr.io/adobe/koperator/jmx-javaagent:1.5.0" # renovate: datasource=docker depName=ghcr.io/adobe/koperator/jmx-javaagent headlessServiceEnabled: true propagateLabels: false 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 + taintedBrokersSelector: + matchLabels: + restart: me readOnlyConfig: | auto.create.topics.enable=false cruise.control.metrics.topic.auto.create=true @@ -30,6 +33,9 @@ spec: requests: storage: 10Gi broker: + kafkaJvmPerfOpts: >- + -Dcom.sun.management.jmxremote.port=1090 -Dcom.sun.management.jmxremote.rmi.port=1090 + -Dcom.sun.management.jmxremote.local.only=false -Djava.rmi.server.hostname=127.0.0.1 -Djute.maxbuffer=0x9fffff processRoles: - broker storageConfigs: @@ -45,27 +51,26 @@ spec: prometheus.io/port: "9020" brokers: - id: 0 - brokerConfigGroup: "broker" - - id: 1 - brokerConfigGroup: "broker" - - id: 2 - brokerConfigGroup: "broker" - - id: 3 brokerConfigGroup: "default" brokerConfig: processRoles: - controller - # - broker - - id: 4 + - id: 1 brokerConfigGroup: "default" brokerConfig: processRoles: - controller - - id: 5 + - id: 2 brokerConfigGroup: "default" brokerConfig: processRoles: - controller + - id: 100 + brokerConfigGroup: "broker" + - id: 101 + brokerConfigGroup: "broker" + - id: 102 + brokerConfigGroup: "broker" rollingUpgradeConfig: failureThreshold: 1 listenersConfig: @@ -300,3 +305,8 @@ spec: { "min.insync.replicas": 3 } + # NOTE: intentionally NO capacityConfig here. A "brokerId": "-1" universal-default entry makes + # GenerateCapacityConfig return early, so Cruise Control's capacity.json stays constant across scaling + # and the #301 capacity-roll path (and its fix) is never exercised. Omitting it forces per-broker + # capacity generation, which is exactly what the KRaft broker-scaling e2e regression-tests. Do not add + # a universal default back or the regression coverage silently disappears. \ No newline at end of file diff --git a/config/samples/simplekafkacluster-with-brokerbindings.yaml b/config/samples/simplekafkacluster-with-brokerbindings.yaml index 4fb0575d8..a1ca0a84d 100644 --- a/config/samples/simplekafkacluster-with-brokerbindings.yaml +++ b/config/samples/simplekafkacluster-with-brokerbindings.yaml @@ -11,7 +11,7 @@ spec: - "zookeeper-server-client.zookeeper:2181" propagateLabels: false 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 diff --git a/config/samples/simplekafkacluster-with-nodeport-external.yaml b/config/samples/simplekafkacluster-with-nodeport-external.yaml index ebd73cac1..2335c0cc2 100644 --- a/config/samples/simplekafkacluster-with-nodeport-external.yaml +++ b/config/samples/simplekafkacluster-with-nodeport-external.yaml @@ -11,7 +11,7 @@ spec: - "zookeeper-server-client.zookeeper:2181" propagateLabels: false 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 diff --git a/config/samples/simplekafkacluster.yaml b/config/samples/simplekafkacluster.yaml index 86320b03b..157619f3d 100644 --- a/config/samples/simplekafkacluster.yaml +++ b/config/samples/simplekafkacluster.yaml @@ -8,13 +8,13 @@ spec: localDebugEnabled: true kRaft: false monitoringConfig: - jmxImage: "ghcr.io/adobe/koperator/jmx-javaagent:1.4.0" + jmxImage: "ghcr.io/adobe/koperator/jmx-javaagent:1.5.0" # renovate: datasource=docker depName=ghcr.io/adobe/koperator/jmx-javaagent headlessServiceEnabled: false zkAddresses: - "zookeeper-server-client.zookeeper:2181" propagateLabels: false 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 @@ -85,7 +85,7 @@ spec: # limits: # cpu: 500m # memory: 1Gi - image: "adobe/cruise-control:3.0.3-adbe-20250804" + image: "adobe/cruise-control:3.0.3-adbe-20260722" # renovate: datasource=docker depName=adobe/cruise-control config: | # Copyright 2017 LinkedIn Corp. Licensed under the BSD 2-Clause License (the "License"). See License in the project root for license information. # diff --git a/config/samples/simplekafkacluster_2disk.yaml b/config/samples/simplekafkacluster_2disk.yaml index 44aa187dd..1b03d6bb8 100644 --- a/config/samples/simplekafkacluster_2disk.yaml +++ b/config/samples/simplekafkacluster_2disk.yaml @@ -7,13 +7,13 @@ metadata: spec: kRaft: false monitoringConfig: - jmxImage: "ghcr.io/adobe/koperator/jmx-javaagent:1.4.0" + jmxImage: "ghcr.io/adobe/koperator/jmx-javaagent:1.5.0" # renovate: datasource=docker depName=ghcr.io/adobe/koperator/jmx-javaagent headlessServiceEnabled: true zkAddresses: - "zookeeper-server-client.zookeeper:2181" propagateLabels: false 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 @@ -91,7 +91,7 @@ spec: # limits: # cpu: 500m # memory: 1Gi - image: "adobe/cruise-control:3.0.3-adbe-20250804" + image: "adobe/cruise-control:3.0.3-adbe-20260722" # renovate: datasource=docker depName=adobe/cruise-control config: | # Copyright 2017 LinkedIn Corp. Licensed under the BSD 2-Clause License (the "License"). See License in the project root for license information. # diff --git a/config/samples/simplekafkacluster_4disk.yaml b/config/samples/simplekafkacluster_4disk.yaml index 3da5b1209..3ec92ffd4 100644 --- a/config/samples/simplekafkacluster_4disk.yaml +++ b/config/samples/simplekafkacluster_4disk.yaml @@ -7,13 +7,13 @@ metadata: spec: kRaft: false monitoringConfig: - jmxImage: "ghcr.io/adobe/koperator/jmx-javaagent:1.4.0" + jmxImage: "ghcr.io/adobe/koperator/jmx-javaagent:1.5.0" # renovate: datasource=docker depName=ghcr.io/adobe/koperator/jmx-javaagent headlessServiceEnabled: true zkAddresses: - "zookeeper-server-client.zookeeper:2181" propagateLabels: false 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 @@ -105,7 +105,7 @@ spec: # limits: # cpu: 500m # memory: 1Gi - image: "adobe/cruise-control:3.0.3-adbe-20250804" + image: "adobe/cruise-control:3.0.3-adbe-20260722" # renovate: datasource=docker depName=adobe/cruise-control config: | # Copyright 2017 LinkedIn Corp. Licensed under the BSD 2-Clause License (the "License"). See License in the project root for license information. # diff --git a/config/samples/simplekafkacluster_5broker.yaml b/config/samples/simplekafkacluster_5broker.yaml index e8409fe19..b9cdb31ce 100644 --- a/config/samples/simplekafkacluster_5broker.yaml +++ b/config/samples/simplekafkacluster_5broker.yaml @@ -7,13 +7,13 @@ metadata: spec: kRaft: false monitoringConfig: - jmxImage: "ghcr.io/adobe/koperator/jmx-javaagent:1.4.0" + jmxImage: "ghcr.io/adobe/koperator/jmx-javaagent:1.5.0" # renovate: datasource=docker depName=ghcr.io/adobe/koperator/jmx-javaagent headlessServiceEnabled: true zkAddresses: - "zookeeper-server-client.zookeeper:2181" propagateLabels: false 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 @@ -88,7 +88,7 @@ spec: # limits: # cpu: 500m # memory: 1Gi - image: "adobe/cruise-control:3.0.3-adbe-20250804" + image: "adobe/cruise-control:3.0.3-adbe-20260722" # renovate: datasource=docker depName=adobe/cruise-control config: | # Copyright 2017 LinkedIn Corp. Licensed under the BSD 2-Clause License (the "License"). See License in the project root for license information. # diff --git a/config/samples/simplekafkacluster_affinity.yaml b/config/samples/simplekafkacluster_affinity.yaml index fd55058e4..b23d5afd6 100644 --- a/config/samples/simplekafkacluster_affinity.yaml +++ b/config/samples/simplekafkacluster_affinity.yaml @@ -13,7 +13,7 @@ spec: # If true OneBrokerPerNode ensures that each kafka broker will be placed on a different node unless a custom # Affinity definition overrides this behavior either set in brokerConfigGroups or in one specific brokerConfig 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 diff --git a/config/samples/simplekafkacluster_ebs_csi.yaml b/config/samples/simplekafkacluster_ebs_csi.yaml index 92af442ce..21c9a67ef 100644 --- a/config/samples/simplekafkacluster_ebs_csi.yaml +++ b/config/samples/simplekafkacluster_ebs_csi.yaml @@ -16,7 +16,7 @@ spec: zkPath: "/kafka" propagateLabels: true 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 diff --git a/config/samples/simplekafkacluster_ssl.yaml b/config/samples/simplekafkacluster_ssl.yaml index 6c19f6a5c..4d7e42c83 100644 --- a/config/samples/simplekafkacluster_ssl.yaml +++ b/config/samples/simplekafkacluster_ssl.yaml @@ -11,7 +11,7 @@ spec: - "zookeeper-server-client.zookeeper:2181" propagateLabels: false 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 diff --git a/config/samples/simplekafkacluster_with_contour.yaml b/config/samples/simplekafkacluster_with_contour.yaml index 14a5db5b3..644147748 100644 --- a/config/samples/simplekafkacluster_with_contour.yaml +++ b/config/samples/simplekafkacluster_with_contour.yaml @@ -6,13 +6,13 @@ metadata: name: kafka spec: monitoringConfig: - jmxImage: "ghcr.io/adobe/koperator/jmx-javaagent:1.4.0" + jmxImage: "ghcr.io/adobe/koperator/jmx-javaagent:1.5.0" # renovate: datasource=docker depName=ghcr.io/adobe/koperator/jmx-javaagent headlessServiceEnabled: true zkAddresses: - "zookeeper-server-client.zookeeper:2181" propagateLabels: false 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 ingressController: "contour" readOnlyConfig: | auto.create.topics.enable=false diff --git a/config/samples/simplekafkacluster_with_k8s_provided_nodeport.yaml b/config/samples/simplekafkacluster_with_k8s_provided_nodeport.yaml index 5013ce406..d30c9d1ee 100644 --- a/config/samples/simplekafkacluster_with_k8s_provided_nodeport.yaml +++ b/config/samples/simplekafkacluster_with_k8s_provided_nodeport.yaml @@ -11,7 +11,7 @@ spec: - "zookeeper-server-client.zookeeper:2181" propagateLabels: false 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 diff --git a/config/samples/simplekafkacluster_with_sasl.yaml b/config/samples/simplekafkacluster_with_sasl.yaml index 2deec9224..572ffd352 100644 --- a/config/samples/simplekafkacluster_with_sasl.yaml +++ b/config/samples/simplekafkacluster_with_sasl.yaml @@ -11,7 +11,7 @@ spec: - "zookeeper-server-client.zookeeper:2181" propagateLabels: false 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 diff --git a/controllers/controller_common.go b/controllers/controller_common.go index c86d8810b..c0d0320a0 100644 --- a/controllers/controller_common.go +++ b/controllers/controller_common.go @@ -84,18 +84,15 @@ func checkBrokerConnectionError(logger logr.Logger, err error) (ctrl.Result, err switch { case errors.As(err, &errorfactory.BrokersUnreachable{}): return ctrl.Result{ - Requeue: true, RequeueAfter: time.Duration(15) * time.Second, }, nil case errors.As(err, &errorfactory.BrokersNotReady{}): return ctrl.Result{ - Requeue: true, RequeueAfter: time.Duration(15) * time.Second, }, nil case errors.As(err, &errorfactory.ResourceNotReady{}): logger.Info("Needed resource for broker connection not found, may not be ready") return ctrl.Result{ - Requeue: true, RequeueAfter: time.Duration(5) * time.Second, }, nil default: diff --git a/controllers/kafkacluster_controller.go b/controllers/kafkacluster_controller.go index 4b825d773..314538703 100644 --- a/controllers/kafkacluster_controller.go +++ b/controllers/kafkacluster_controller.go @@ -17,6 +17,7 @@ package controllers import ( "context" + "encoding/json" "fmt" "reflect" "time" @@ -268,7 +269,6 @@ func (r *KafkaClusterReconciler) checkFinalizers(ctx context.Context, cluster *v if len(childTopics.Items) > 0 { log.Info(fmt.Sprintf("Still waiting for the following topics to be deleted: %v", topicListToStrSlice(childTopics))) return ctrl.Result{ - Requeue: true, RequeueAfter: time.Duration(3) * time.Second, }, nil } @@ -303,7 +303,6 @@ func (r *KafkaClusterReconciler) checkFinalizers(ctx context.Context, cluster *v if errors.As(err, &errorfactory.ResourceNotReady{}) { log.Info("The PKI is not ready to be torn down") return ctrl.Result{ - Requeue: true, RequeueAfter: time.Duration(5) * time.Second, }, nil } @@ -359,6 +358,41 @@ func (r *KafkaClusterReconciler) updateAndFetchLatest(ctx context.Context, clust return cluster, nil } +// ignoreMetadataFields returns a patch.CalculateOption that drops only the given fields +// under metadata (e.g. "managedFields", "resourceVersion") instead of the whole metadata +// object like patch.IgnoreField("metadata") does, so other metadata changes (labels, +// annotations, etc.) still count toward the diff. +func ignoreMetadataFields(fields ...string) patch.CalculateOption { + return func(current, modified []byte) ([]byte, []byte, error) { + current, err := deleteMetadataFields(current, fields...) + if err != nil { + return nil, nil, errors.WrapIf(err, "could not delete metadata fields from current byte sequence") + } + + modified, err = deleteMetadataFields(modified, fields...) + if err != nil { + return nil, nil, errors.WrapIf(err, "could not delete metadata fields from modified byte sequence") + } + + return current, modified, nil + } +} + +func deleteMetadataFields(obj []byte, fields ...string) ([]byte, error) { + var objectMap map[string]interface{} + if err := json.Unmarshal(obj, &objectMap); err != nil { + return nil, errors.WrapIf(err, "could not unmarshal byte sequence") + } + + if metadata, ok := objectMap["metadata"].(map[string]interface{}); ok { + for _, field := range fields { + delete(metadata, field) + } + } + + return json.Marshal(objectMap) +} + // SetupKafkaClusterWithManager registers kafka cluster controller to the manager func SetupKafkaClusterWithManager(mgr ctrl.Manager, contourEnabled bool) *ctrl.Builder { log := mgr.GetLogger() @@ -390,7 +424,7 @@ func SetupKafkaClusterWithManager(mgr ctrl.Manager, contourEnabled bool) *ctrl.B UpdateFunc: func(e event.UpdateEvent) bool { switch newObj := e.ObjectNew.(type) { case *corev1.Pod, *corev1.ConfigMap, *corev1.PersistentVolumeClaim: - patchResult, err := patch.DefaultPatchMaker.Calculate(e.ObjectOld, e.ObjectNew) + patchResult, err := patch.DefaultPatchMaker.Calculate(e.ObjectOld, e.ObjectNew, patch.IgnoreStatusFields(), ignoreMetadataFields("managedFields", "resourceVersion")) if err != nil { log.Error(err, "could not match objects", "kind", e.ObjectOld.GetObjectKind()) } else if patchResult.IsEmpty() { diff --git a/controllers/kafkauser_controller.go b/controllers/kafkauser_controller.go index 243347a74..54c67774d 100644 --- a/controllers/kafkauser_controller.go +++ b/controllers/kafkauser_controller.go @@ -224,7 +224,6 @@ func (r *KafkaUserReconciler) Reconcile(ctx context.Context, request reconcile.R case errors.As(err, &errorfactory.ResourceNotReady{}): reqLogger.Info("generated secret not found, may not be ready") return ctrl.Result{ - Requeue: true, RequeueAfter: time.Duration(5) * time.Second, }, nil case errors.As(err, &errorfactory.FatalReconcileError{}): @@ -233,7 +232,6 @@ func (r *KafkaUserReconciler) Reconcile(ctx context.Context, request reconcile.R // The user can fix while this is looping and it will pick it up next reconcile attempt reqLogger.Error(err, "Fatal error attempting to reconcile the user certificate.") return ctrl.Result{ - Requeue: true, RequeueAfter: time.Duration(15) * time.Second, }, nil default: @@ -250,9 +248,7 @@ func (r *KafkaUserReconciler) Reconcile(ctx context.Context, request reconcile.R kafkaUser, err = user.GetDistinguishedName() if err != nil { reqLogger.Error(err, "could not get Distinguished Name from the generated TLS certificate", "cert", string(user.Certificate)) - return ctrl.Result{ - Requeue: false, - }, err + return ctrl.Result{}, err } // check if marked for deletion and remove created certs if k8sutil.IsMarkedForDeletion(instance.ObjectMeta) { diff --git a/docker/jmx_exporter/Dockerfile b/docker/jmx_exporter/Dockerfile index 0e31b22af..20e0d5cf9 100644 --- a/docker/jmx_exporter/Dockerfile +++ b/docker/jmx_exporter/Dockerfile @@ -1,6 +1,6 @@ ARG JMX_EXPORTER_VERSION=v1.6.0 # renovate: datasource=github-releases depName=prometheus/jmx_exporter -FROM maven:3-amazoncorretto-21@sha256:a7f84ee223e9b6a1ba462667cdb98e282adf79d02fbd1ad3091ad0bb2feff1d7 AS build +FROM maven:3-amazoncorretto-21@sha256:ae694e8f35dfdb3f74f70f5f5cb3290d64fc7179f11cd9f0d493851c024aec77 AS build ARG JMX_EXPORTER_VERSION # Install wget to download the release tarball diff --git a/docs/benchmarks/infrastructure/kafka.yaml b/docs/benchmarks/infrastructure/kafka.yaml index 9e9e536f0..7ce9230df 100644 --- a/docs/benchmarks/infrastructure/kafka.yaml +++ b/docs/benchmarks/infrastructure/kafka.yaml @@ -10,7 +10,7 @@ spec: zkAddresses: - "zookeeper-server-client.zookeeper:2181" oneBrokerPerNode: true - 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 rackAwareness: diff --git a/docs/benchmarks/infrastructure/kafka_202001_3brokers.yaml b/docs/benchmarks/infrastructure/kafka_202001_3brokers.yaml index d8941a016..e21eb14ae 100644 --- a/docs/benchmarks/infrastructure/kafka_202001_3brokers.yaml +++ b/docs/benchmarks/infrastructure/kafka_202001_3brokers.yaml @@ -7,7 +7,7 @@ metadata: spec: kRaft: false monitoringConfig: - jmxImage: "ghcr.io/adobe/koperator/jmx-javaagent:1.5.0" + jmxImage: "ghcr.io/adobe/koperator/jmx-javaagent:1.5.0" # renovate: datasource=docker depName=ghcr.io/adobe/koperator/jmx-javaagent pathToJar: "/jmx_prometheus_javaagent.jar" kafkaJMXExporterConfig: |2 lowercaseOutputName: true @@ -66,7 +66,7 @@ spec: zkAddresses: - "zookeeper-server-client.zookeeper:2181" oneBrokerPerNode: true - 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 min.insync.replicas=3 diff --git a/docs/examples/springboot-kafka-avro/Dockerfile b/docs/examples/springboot-kafka-avro/Dockerfile index 55cbdc2f1..5caafe5c0 100644 --- a/docs/examples/springboot-kafka-avro/Dockerfile +++ b/docs/examples/springboot-kafka-avro/Dockerfile @@ -1,4 +1,4 @@ -FROM maven:3.9-eclipse-temurin-21-alpine@sha256:d6390509b774df0a07d009a194a020aeff9c715afe1e3d2cd5ecb3bf6344700f as build +FROM maven:3.9-eclipse-temurin-21-alpine@sha256:d9690b76996b8117806b012c57883bbdcbe8c48fca0636166eeeeeff6e8e8afe as build # Set working directory WORKDIR /usr/src/myapp diff --git a/docs/examples/springboot-kafka-avro/pom.xml b/docs/examples/springboot-kafka-avro/pom.xml index cca93123f..b53f0d035 100644 --- a/docs/examples/springboot-kafka-avro/pom.xml +++ b/docs/examples/springboot-kafka-avro/pom.xml @@ -51,7 +51,7 @@ org.apache.avro avro - 1.12.1 + 1.12.2 io.confluent @@ -70,7 +70,7 @@ org.apache.avro avro-maven-plugin - 1.12.1 + 1.12.2 generate-sources diff --git a/hack/kafka-test-pod/Dockerfile b/hack/kafka-test-pod/Dockerfile index d7eff6ef3..222b83b36 100644 --- a/hack/kafka-test-pod/Dockerfile +++ b/hack/kafka-test-pod/Dockerfile @@ -1,4 +1,4 @@ -FROM golang:1.26@sha256:2005724102f45917a63e9d092fc0e4ea56ea575048ce147caad5f5f61502c365 as builder +FROM golang:1.27@sha256:65b6f280bf050ec5af12716857e8ea8439d694dbba8f31ceeb7630670071f2bb as builder WORKDIR /workspace diff --git a/hack/kafka-test-pod/manifest_certmanager.yaml b/hack/kafka-test-pod/manifest_certmanager.yaml index 584e18fba..302c65dbc 100644 --- a/hack/kafka-test-pod/manifest_certmanager.yaml +++ b/hack/kafka-test-pod/manifest_certmanager.yaml @@ -66,7 +66,7 @@ spec: # Container reading from topic with the consumer credentials - name: consumer - image: banzaicloud/kafka-test:latest + image: banzaicloud/kafka-test:latest # renovate: datasource=docker depName=banzaicloud/kafka-test imagePullPolicy: IfNotPresent env: - name: KAFKA_MODE @@ -78,7 +78,7 @@ spec: # Container writing to topic with the producer credentials - name: producer - image: banzaicloud/kafka-test:latest + image: banzaicloud/kafka-test:latest # renovate: datasource=docker depName=banzaicloud/kafka-test imagePullPolicy: IfNotPresent env: - name: KAFKA_MODE diff --git a/pkg/jmxextractor/extractor.go b/pkg/jmxextractor/extractor.go index 583e06dd7..099fdcdb6 100644 --- a/pkg/jmxextractor/extractor.go +++ b/pkg/jmxextractor/extractor.go @@ -20,6 +20,7 @@ import ( "io" "net/http" "regexp" + "time" "github.com/banzaicloud/koperator/api/v1beta1" "github.com/banzaicloud/koperator/pkg/errorfactory" @@ -30,9 +31,9 @@ import ( ) const ( - headlessServiceJMXTemplate = "http://%s-%d." + kafka.HeadlessServiceTemplate + ".%s.svc.%s:%d" - headlessControllerServiceJMXTemplate = "http://%s-%d." + kafka.HeadlessControllerServiceTemplate + ".%s.svc.%s:%d" - serviceJMXTemplate = "http://%s-%d.%s.svc.%s:%d" + headlessServiceJMXTemplate = "http://%s-%d." + kafka.HeadlessServiceTemplate + ".%s.svc.%s:%d/metrics" + headlessControllerServiceJMXTemplate = "http://%s-%d." + kafka.HeadlessControllerServiceTemplate + ".%s.svc.%s:%d/metrics" + serviceJMXTemplate = "http://%s-%d.%s.svc.%s:%d/metrics" versionRegexGroup = "version" ) @@ -90,7 +91,8 @@ func (exp *jmxExtractor) ExtractDockerImageAndVersion(brokerId int32, brokerConf requestURL = fmt.Sprintf(serviceJMXTemplate, exp.clusterName, brokerId, exp.clusterNamespace, exp.kubernetesClusterDomain, 9020) } - rsp, err := http.Get(requestURL) + client := &http.Client{Timeout: 30 * time.Second} + rsp, err := client.Get(requestURL) if err != nil { exp.log.Error(err, fmt.Sprintf("error during talking to broker-%d", brokerId)) return nil, errorfactory.New(errorfactory.BrokersNotReady{}, err, "unable to talk to ...") diff --git a/pkg/k8sutil/status.go b/pkg/k8sutil/status.go index 4836cb3ab..de8b36194 100644 --- a/pkg/k8sutil/status.go +++ b/pkg/k8sutil/status.go @@ -127,7 +127,7 @@ func UpdateBrokerStatus(c client.Client, brokerIDs []string, cluster *banzaiclou } // update loses the typeMeta of the config that's used later when setting ownerrefs cluster.TypeMeta = typeMeta - logger.Info("Kafka cluster state updated") + logger.V(1).Info("Broker status updated", "brokers", brokerIDs, "status", state) return nil } diff --git a/pkg/kafkaclient/client.go b/pkg/kafkaclient/client.go index 83f388596..8f0a23d68 100644 --- a/pkg/kafkaclient/client.go +++ b/pkg/kafkaclient/client.go @@ -130,7 +130,7 @@ func NewFromCluster(k8sclient client.Client, cluster *v1beta1.KafkaCluster) (Kaf if err := client.Close(); err != nil { log.Error(err, "Error closing Kafka client") } else { - log.Info("Kafka client closed cleanly") + log.V(1).Info("Kafka client closed cleanly") } } return client, close, err diff --git a/pkg/resources/cruisecontrol/configmap.go b/pkg/resources/cruisecontrol/configmap.go index 20735de89..b121f70dc 100644 --- a/pkg/resources/cruisecontrol/configmap.go +++ b/pkg/resources/cruisecontrol/configmap.go @@ -157,7 +157,7 @@ type JBODInvariantCapacityConfig struct { func GenerateCapacityConfig(kafkaCluster *v1beta1.KafkaCluster, log logr.Logger, config *corev1.ConfigMap) (string, error) { var err error - log.Info("generating capacity config") + log.V(2).Info("generating capacity config") var capacityConfig JBODInvariantCapacityConfig var userConfigBrokerIds []string @@ -184,7 +184,7 @@ func GenerateCapacityConfig(kafkaCluster *v1beta1.KafkaCluster, log logr.Logger, } // If the -1 default exists we don't have to do anything else here since all brokers will have values. if brokerId == "-1" { - log.Info("Using user provided capacity config because it has universal default defined", "capacity config", userProvidedCapacityConfig) + log.V(2).Info("Using user provided capacity config because it has universal default defined", "capacity config", userProvidedCapacityConfig) return userProvidedCapacityConfig, nil } userConfigBrokerIds = append(userConfigBrokerIds, brokerId) @@ -211,7 +211,7 @@ func GenerateCapacityConfig(kafkaCluster *v1beta1.KafkaCluster, log logr.Logger, if err != nil { return "", errors.WrapIf(err, "could not marshal cruise control capacity config") } - log.Info("broker capacity config generated successfully", "capacity config", string(result)) + log.V(2).Info("broker capacity config generated successfully", "capacity config", string(result)) return string(result), err } diff --git a/pkg/resources/kafka/allBrokerService.go b/pkg/resources/kafka/allBrokerService.go index b5fa40239..8c83dd417 100644 --- a/pkg/resources/kafka/allBrokerService.go +++ b/pkg/resources/kafka/allBrokerService.go @@ -46,10 +46,11 @@ func (r *Reconciler) allBrokerService() runtime.Object { r.KafkaCluster.Spec.ListenersConfig.GetServiceAnnotations(), r.KafkaCluster), Spec: corev1.ServiceSpec{ - Type: corev1.ServiceTypeClusterIP, - SessionAffinity: corev1.ServiceAffinityNone, - Selector: apiutil.LabelsForKafka(r.KafkaCluster.GetName()), - Ports: usedPorts, + Type: corev1.ServiceTypeClusterIP, + SessionAffinity: corev1.ServiceAffinityNone, + Selector: apiutil.LabelsForKafka(r.KafkaCluster.GetName()), + Ports: usedPorts, + PublishNotReadyAddresses: r.KafkaCluster.Spec.PublishNotReadyAddresses, }, } diff --git a/pkg/resources/kafka/kafka.go b/pkg/resources/kafka/kafka.go index d737754f8..935b51577 100644 --- a/pkg/resources/kafka/kafka.go +++ b/pkg/resources/kafka/kafka.go @@ -211,7 +211,7 @@ func (r *Reconciler) Reconcile(log logr.Logger) error { log.V(1).Info("Reconciling") - log.Info("broker rack map", "kafkaBrokerAvailabilityZoneMap", getBrokerAzMap(r.KafkaCluster)) + log.V(1).Info("broker rack map", "kafkaBrokerAvailabilityZoneMap", getBrokerAzMap(r.KafkaCluster)) ctx := context.Background() if err := k8sutil.UpdateBrokerConfigurationBackup(r.Client, r.KafkaCluster); err != nil { @@ -414,6 +414,7 @@ func (r *Reconciler) Reconcile(log logr.Logger) error { reorderedBrokers := reorderBrokers(runningBrokers, boundPersistentVolumeClaims, r.KafkaCluster.Spec.Brokers, r.KafkaCluster.Status.BrokersState, controllerID, log) allBrokerDynamicConfigSucceeded := true + brokerStatus := make(map[int32]*banzaiv1beta1.BrokerConfig) for _, broker := range reorderedBrokers { brokerConfig, err := broker.GetBrokerConfig(r.KafkaCluster.Spec) if err != nil { @@ -454,9 +455,10 @@ func (r *Reconciler) Reconcile(log logr.Logger) error { if err != nil { return err } - if err = r.updateStatusWithDockerImageAndVersion(broker.Id, brokerConfig, log); err != nil { - return err + if r.brokerNeedsVersionUpdate(broker.Id, brokerConfig) { + brokerStatus[broker.Id] = brokerConfig } + // If dynamic configs can not be set then let the loop continue to the next broker, // after the loop we return error. This solves that case when other brokers could get healthy, // but the loop exits too soon because dynamic configs can not be set. @@ -479,11 +481,31 @@ func (r *Reconciler) Reconcile(log logr.Logger) error { return err } + if err := r.updateStatusWithDockerImageAndVersion(brokerStatus, log); err != nil { + return err + } + // in case HeadlessServiceEnabled is changed, delete the service that was created by the previous // reconcile flow. The services must be deleted at the end of the reconcile flow after the new services // were created and broker configurations reflecting the new services otherwise the Kafka brokers // won't be reachable by koperator. - if r.KafkaCluster.Spec.HeadlessServiceEnabled { + // + // A broker pod can still be running outside spec.Brokers while its removal is pending. Deletion is gated + // on Cruise Control finishing the data migration off it. + // That pod is not touched by the per-broker loop above, so it never gets a replacement address on the + // other addressing scheme; its only reachable address is whichever Service this reconcile is about to + // delete. Deleting it out from under a broker that is still alive and still needed would orphan that + // broker from the network entirely, stalling the very Cruise Control operation the removal is waiting on. + // Rather than reasoning about whether this particular deletion happens to be safe this time, + // skip both service-topology transitions while any such pod exists, log it clearly, and let a later + // reconcile retry once the pod is gone. + if pendingBrokerID, stillPending := firstRunningBrokerOutsideSpec(runningBrokers, r.KafkaCluster.Spec.Brokers); stillPending { + log.Info("deferring headless/non-headless service reconciliation: a broker outside spec.Brokers still has a running pod", + "component", componentName, + "clusterName", r.KafkaCluster.Name, + "clusterNamespace", r.KafkaCluster.Namespace, + "pendingBrokerId", pendingBrokerID) + } else if r.KafkaCluster.Spec.HeadlessServiceEnabled { log.V(1).Info("deleting non-headless services for all of the brokers") if err := r.deleteNonHeadlessServices(ctx); err != nil { @@ -602,8 +624,10 @@ func (r *Reconciler) reconcileKafkaPodDelete(ctx context.Context, log logr.Logge } processRoles, found := broker.GetLabels()[banzaiv1beta1.ProcessRolesKey] - // only applicable in KRaft: if this Kafka pod is not a controller-only node, there is no corresponding CC - // therefore we can just skip the broker state check and delete the pod safely + // Only applicable in KRaft: a controller-only node has no corresponding Cruise Control broker, + // so it has no graceful-downscale state to honor and can be deleted directly (this branch is skipped). + // For every other node (broker-only, combined, or non-KRaft) we wait for the graceful downscale + // to finish before deleting the pod. if !found || processRoles != banzaiv1beta1.ControllerNodeProcessRole { if brokerState, ok := r.KafkaCluster.Status.BrokersState[broker.Labels[banzaiv1beta1.BrokerIdLabelKey]]; ok && brokerState.GracefulActionState.CruiseControlState != banzaiv1beta1.GracefulDownscaleSucceeded && @@ -898,7 +922,7 @@ func (r *Reconciler) reconcileKafkaPod(log logr.Logger, desiredPod *corev1.Pod, brokerId := currentPod.Labels[banzaiv1beta1.BrokerIdLabelKey] if _, ok := r.KafkaCluster.Status.BrokersState[brokerId]; ok { if currentPod.Spec.NodeName == "" { - log.Info(fmt.Sprintf("pod for brokerId %s does not scheduled to node yet", brokerId)) + log.V(1).Info(fmt.Sprintf("pod for brokerId %s not scheduled to node yet", brokerId)) } else if r.KafkaCluster.Spec.RackAwareness != nil { rackAwarenessState, err := k8sutil.UpdateCrWithRackAwarenessConfig(currentPod, r.KafkaCluster, r.Client, r.DirectClient) if err != nil { @@ -922,21 +946,44 @@ func (r *Reconciler) reconcileKafkaPod(log logr.Logger, desiredPod *corev1.Pod, return nil } -func (r *Reconciler) updateStatusWithDockerImageAndVersion(brokerId int32, brokerConfig *banzaiv1beta1.BrokerConfig, - log logr.Logger) error { - jmxExp := jmxextractor.NewJMXExtractor(r.KafkaCluster.GetNamespace(), - r.KafkaCluster.Spec.GetKubernetesClusterDomain(), r.KafkaCluster.GetName(), log) +// brokerNeedsVersionUpdate returns true when the broker's image/version status is absent, +// incomplete, or stale relative to the desired image — i.e. a JMX fetch is warranted. +func (r *Reconciler) brokerNeedsVersionUpdate(brokerID int32, brokerConfig *banzaiv1beta1.BrokerConfig) bool { + desiredImage := util.GetBrokerImage(brokerConfig, r.KafkaCluster.Spec.GetClusterImage()) + state, ok := r.KafkaCluster.Status.BrokersState[strconv.Itoa(int(brokerID))] + return !ok || state.Version == "" || state.Image != desiredImage +} - kafkaVersion, err := jmxExp.ExtractDockerImageAndVersion(brokerId, brokerConfig, - r.KafkaCluster.Spec.GetClusterImage(), r.KafkaCluster.Spec.HeadlessServiceEnabled) - if err != nil { - return err +type brokerVersionResult struct { + brokerID int32 + kafkaVersion *banzaiv1beta1.KafkaVersion + err error +} + +func (r *Reconciler) updateStatusWithDockerImageAndVersion(brokers map[int32]*banzaiv1beta1.BrokerConfig, log logr.Logger) error { + ch := make(chan brokerVersionResult, len(brokers)) + for brokerID, brokerConfig := range brokers { + go func(id int32, cfg *banzaiv1beta1.BrokerConfig) { + jmxExp := jmxextractor.NewJMXExtractor(r.KafkaCluster.GetNamespace(), r.KafkaCluster.Spec.GetKubernetesClusterDomain(), r.KafkaCluster.GetName(), log) + kv, err := jmxExp.ExtractDockerImageAndVersion(id, cfg, r.KafkaCluster.Spec.GetClusterImage(), r.KafkaCluster.Spec.HeadlessServiceEnabled) + if err != nil { + ch <- brokerVersionResult{brokerID: id, err: err} + return + } + ch <- brokerVersionResult{brokerID: id, kafkaVersion: kv} + }(brokerID, brokerConfig) } - err = k8sutil.UpdateBrokerStatus(r.Client, []string{strconv.Itoa(int(brokerId))}, r.KafkaCluster, - *kafkaVersion, log) - if err != nil { - return err + + for range brokers { + result := <-ch + if result.err != nil { + return result.err + } + if err := k8sutil.UpdateBrokerStatus(r.Client, []string{strconv.Itoa(int(result.brokerID))}, r.KafkaCluster, *result.kafkaVersion, log); err != nil { + return err + } } + return nil } @@ -962,7 +1009,7 @@ func (r *Reconciler) handleRollingUpgrade(log logr.Logger, desiredPod, currentPo case err != nil: log.Error(err, "could not match objects", "kind", desiredType) case r.isPodTainted(log, currentPod): - log.Info("pod has tainted labels, deleting it", "pod", currentPod) + log.Info("pod has tainted labels, attempting to delete", "pod", currentPod) case patchResult.IsEmpty(): if !k8sutil.IsPodContainsTerminatedContainer(currentPod) && r.KafkaCluster.Status.BrokersState[currentPod.Labels[banzaiv1beta1.BrokerIdLabelKey]].ConfigurationState == banzaiv1beta1.ConfigInSync && @@ -1003,6 +1050,20 @@ func (r *Reconciler) handleRollingUpgrade(log logr.Logger, desiredPod, currentPo return errorfactory.New(errorfactory.ReconcileRollingUpgrade{}, errors.New("pod count differs from brokers spec"), "rolling upgrade in progress") } + // In KRaft mode, do not delete a controller pod while any other controller is still catching up + // in the metadata quorum. A controller pod's readiness reflects quorum membership (see its + // readiness probe), so waiting for all other controllers to be Ready keeps the operator from + // restarting the next controller before the previously restarted one has rejoined the quorum - + // which could otherwise cost the quorum its majority. The pod being reconciled is excluded so an + // unhealthy controller can still be replaced. Broker-only restarts cannot affect quorum majority + // and are gated by the data-plane health check below, so they are not blocked here. + if notReadyControllers := controllersBlockingRollingUpgrade(r.KafkaCluster.Spec.KRaftMode, podList.Items, currentPod); len(notReadyControllers) > 0 { + return errorfactory.New(errorfactory.ReconcileRollingUpgrade{}, + errors.New("KRaft controller quorum is not stable yet"), + "waiting for KRaft controllers to rejoin the quorum before continuing rolling upgrade", + "notReadyControllerBrokerIDs", strings.Join(notReadyControllers, ",")) + } + // Check if we support multiple broker restarts and restart only in same AZ, otherwise restart only 1 broker at once terminatingOrPendingPods := getPodsInTerminatingOrPendingState(podList.Items) if len(terminatingOrPendingPods) >= r.KafkaCluster.Spec.RollingUpgradeConfig.ConcurrentBrokerRestartCountPerRack { @@ -1032,14 +1093,14 @@ func (r *Reconciler) handleRollingUpgrade(log logr.Logger, desiredPod, currentPo return errors.WrapIf(err, "health check failed") } if len(allOfflineReplicas) > 0 { - log.Info("offline replicas", "IDs", allOfflineReplicas) + log.V(1).Info("offline replicas", "IDs", allOfflineReplicas) } outOfSyncReplicas, err := kClient.OutOfSyncReplicas() if err != nil { return errors.WrapIf(err, "health check failed") } if len(outOfSyncReplicas) > 0 { - log.Info("out-of-sync replicas", "IDs", outOfSyncReplicas) + log.V(1).Info("out-of-sync replicas", "IDs", outOfSyncReplicas) } impactedReplicas := make(map[int32]struct{}) for _, brokerID := range allOfflineReplicas { @@ -1613,6 +1674,50 @@ func getPodsInTerminatingOrPendingState(items []corev1.Pod) []corev1.Pod { return pods } +// controllersBlockingRollingUpgrade returns the not-Ready controller broker IDs that must block the +// rolling upgrade before currentPod is deleted, or nil when the upgrade may proceed. The quorum-readiness +// gate applies only when the cluster is in KRaft mode AND currentPod is itself a controller node: +// restarting a broker-only node cannot cost the metadata quorum its majority (brokers are not voters), +// so broker restarts are gated by the data-plane health check instead, not by controller readiness. +func controllersBlockingRollingUpgrade(kRaftMode bool, pods []corev1.Pod, currentPod *corev1.Pod) []string { + if !kRaftMode || currentPod.Labels[banzaiv1beta1.IsControllerNodeKey] != configValueTrue { + return nil + } + return notReadyControllerBrokerIDs(pods, currentPod) +} + +// notReadyControllerBrokerIDs returns the broker IDs of KRaft controller pods (broker id label of pods +// labeled isControllerNode=true) that are not Ready, excluding the pod currently being reconciled. +// A controller pod's readiness reflects its metadata-quorum membership, so a non-empty result means the +// quorum is still stabilizing and the rolling upgrade should pause before deleting another pod. +func notReadyControllerBrokerIDs(pods []corev1.Pod, currentPod *corev1.Pod) []string { + var notReady []string + currentBrokerID := currentPod.Labels[banzaiv1beta1.BrokerIdLabelKey] + for i := range pods { + pod := &pods[i] + if pod.Labels[banzaiv1beta1.BrokerIdLabelKey] == currentBrokerID { + continue + } + if pod.Labels[banzaiv1beta1.IsControllerNodeKey] != configValueTrue { + continue + } + if !isPodReady(pod) { + notReady = append(notReady, pod.Labels[banzaiv1beta1.BrokerIdLabelKey]) + } + } + return notReady +} + +// isPodReady reports whether the pod's Ready condition is true. +func isPodReady(pod *corev1.Pod) bool { + for _, cond := range pod.Status.Conditions { + if cond.Type == corev1.PodReady { + return cond.Status == corev1.ConditionTrue + } + } + return false +} + func (r *Reconciler) getBrokerAz(pod *corev1.Pod, kafkaBrokerAvailabilityZoneMap map[int32]string) (string, error) { brokerId, err := strconv.ParseInt(pod.Labels[banzaiv1beta1.BrokerIdLabelKey], 10, 32) if err != nil { @@ -1651,6 +1756,25 @@ func getServiceFromExternalListener(client client.Client, cluster *banzaiv1beta1 return foundLBService, nil } +// firstRunningBrokerOutsideSpec reports whether any broker with a running pod (runningBrokers, keyed by +// BrokerIdLabelKey value) is absent from desiredBrokers, returning its ID for logging. Such a broker is +// outside the per-broker reconcile loop entirely (see reorderBrokers) - e.g. one dropped from spec.Brokers +// while its removal is still pending Cruise Control's data migration - so it never receives a replacement +// address on the other Service-addressing scheme. Iteration order over the map is nondeterministic, but +// finding any single one is enough for callers that only need to know whether it is safe to proceed. +func firstRunningBrokerOutsideSpec(runningBrokers map[string]struct{}, desiredBrokers []banzaiv1beta1.Broker) (string, bool) { + inSpec := make(map[string]struct{}, len(desiredBrokers)) + for _, b := range desiredBrokers { + inSpec[strconv.Itoa(int(b.Id))] = struct{}{} + } + for id := range runningBrokers { + if _, ok := inSpec[id]; !ok { + return id, true + } + } + return "", false +} + // reorderBrokers returns the KafkaCluster brokers list reordered for reconciliation such that: // - the controller broker is reconciled last // - prioritize missing broker pods where downscale operation has not been finished yet to give bigger chance to be scheduled and downscale operation to be continued diff --git a/pkg/resources/kafka/kafka_test.go b/pkg/resources/kafka/kafka_test.go index ad9e6db4b..be38350cd 100644 --- a/pkg/resources/kafka/kafka_test.go +++ b/pkg/resources/kafka/kafka_test.go @@ -591,6 +591,52 @@ func TestReorderBrokers(t *testing.T) { } } +func TestFirstRunningBrokerOutsideSpec(t *testing.T) { + testCases := []struct { + testName string + runningBrokers map[string]struct{} + desiredBrokers []v1beta1.Broker + wantFound bool + wantID string + }{ + { + testName: "no running brokers at all", + runningBrokers: map[string]struct{}{}, + desiredBrokers: []v1beta1.Broker{{Id: 0}, {Id: 1}}, + wantFound: false, + }, + { + testName: "every running broker is still in spec", + runningBrokers: map[string]struct{}{"0": {}, "1": {}, "2": {}}, + desiredBrokers: []v1beta1.Broker{{Id: 0}, {Id: 1}, {Id: 2}}, + wantFound: false, + }, + { + testName: "a broker dropped from spec still has a running pod", + // Mirrors the service-transition failure mode: broker 3 was removed from spec.Brokers while its + // pod, pending Cruise Control's data migration, is still Running. + runningBrokers: map[string]struct{}{"0": {}, "1": {}, "2": {}, "3": {}}, + desiredBrokers: []v1beta1.Broker{{Id: 0}, {Id: 1}, {Id: 2}}, + wantFound: true, + wantID: "3", + }, + { + testName: "a desired broker has no running pod yet (scale-up in progress)", + runningBrokers: map[string]struct{}{"0": {}}, + desiredBrokers: []v1beta1.Broker{{Id: 0}, {Id: 1}}, + wantFound: false, + }, + } + + for _, tt := range testCases { + t.Run(tt.testName, func(t *testing.T) { + id, found := firstRunningBrokerOutsideSpec(tt.runningBrokers, tt.desiredBrokers) + assert.Equal(t, tt.wantFound, found) + assert.Equal(t, tt.wantID, id) + }) + } +} + func TestGetServerPasswordKeysAndUsers(t *testing.T) { //nolint funlen t.Parallel() testCases := []struct { @@ -1986,3 +2032,253 @@ func TestGetBrokerAzMap(t *testing.T) { }) } } + +func TestBrokerNeedsVersionUpdate(t *testing.T) { + t.Parallel() + const clusterImage = "apache/kafka:3.4.0" + const updatedImage = "apache/kafka:3.5.0" + + testCases := []struct { + testName string + brokerID int32 + brokerConfig v1beta1.BrokerConfig + clusterImage string + brokersState map[string]v1beta1.BrokerState + expected bool + }{ + { + testName: "no existing status entry triggers update", + brokerID: 0, + brokerConfig: v1beta1.BrokerConfig{}, + clusterImage: clusterImage, + brokersState: map[string]v1beta1.BrokerState{}, + expected: true, + }, + { + testName: "status entry with empty version triggers update", + brokerID: 0, + brokerConfig: v1beta1.BrokerConfig{}, + clusterImage: clusterImage, + brokersState: map[string]v1beta1.BrokerState{ + "0": {Image: clusterImage, Version: ""}, + }, + expected: true, + }, + { + testName: "status entry with different image triggers update", + brokerID: 0, + brokerConfig: v1beta1.BrokerConfig{}, + clusterImage: updatedImage, + brokersState: map[string]v1beta1.BrokerState{ + "0": {Image: clusterImage, Version: "3.4.0"}, + }, + expected: true, + }, + { + testName: "status up to date with cluster image skips update", + brokerID: 0, + brokerConfig: v1beta1.BrokerConfig{}, + clusterImage: clusterImage, + brokersState: map[string]v1beta1.BrokerState{ + "0": {Image: clusterImage, Version: "3.4.0"}, + }, + expected: false, + }, + { + testName: "broker-level image override used instead of cluster image", + brokerID: 0, + brokerConfig: v1beta1.BrokerConfig{Image: "apache/kafka:3.4.1"}, + clusterImage: clusterImage, + brokersState: map[string]v1beta1.BrokerState{ + "0": {Image: "apache/kafka:3.4.1", Version: "3.4.1"}, + }, + expected: false, + }, + { + testName: "broker-level image override differs from recorded image triggers update", + brokerID: 0, + brokerConfig: v1beta1.BrokerConfig{Image: "apache/kafka:3.5.0"}, + clusterImage: clusterImage, + brokersState: map[string]v1beta1.BrokerState{ + "0": {Image: "apache/kafka:3.4.1", Version: "3.4.1"}, + }, + expected: true, + }, + { + testName: "correct state for one broker does not suppress update for another", + brokerID: 1, + brokerConfig: v1beta1.BrokerConfig{}, + clusterImage: clusterImage, + brokersState: map[string]v1beta1.BrokerState{ + "0": {Image: clusterImage, Version: "3.4.0"}, + }, + expected: true, + }, + } + + for _, test := range testCases { + test := test + t.Run(test.testName, func(t *testing.T) { + t.Parallel() + r := Reconciler{ + Reconciler: resources.Reconciler{ + KafkaCluster: &v1beta1.KafkaCluster{ + Spec: v1beta1.KafkaClusterSpec{ + ClusterImage: test.clusterImage, + }, + Status: v1beta1.KafkaClusterStatus{ + BrokersState: test.brokersState, + }, + }, + }, + } + assert.Equal(t, test.expected, r.brokerNeedsVersionUpdate(test.brokerID, &test.brokerConfig)) + }) + } +} + +func TestNotReadyControllerBrokerIDs(t *testing.T) { + makePod := func(brokerID, isController string, ready bool) corev1.Pod { + readyStatus := corev1.ConditionFalse + if ready { + readyStatus = corev1.ConditionTrue + } + return corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{ + Labels: map[string]string{ + v1beta1.BrokerIdLabelKey: brokerID, + v1beta1.IsControllerNodeKey: isController, + }, + }, + Status: corev1.PodStatus{ + Conditions: []corev1.PodCondition{{Type: corev1.PodReady, Status: readyStatus}}, + }, + } + } + + tests := []struct { + testName string + pods []corev1.Pod + currentBroker string + expected []string + }{ + { + testName: "all controllers ready", + pods: []corev1.Pod{ + makePod("0", "false", true), + makePod("1", "true", true), + makePod("2", "true", true), + }, + currentBroker: "0", + expected: nil, + }, + { + testName: "one other controller not ready is reported", + pods: []corev1.Pod{ + makePod("0", "false", true), + makePod("1", "true", false), + makePod("2", "true", true), + }, + currentBroker: "0", + expected: []string{"1"}, + }, + { + testName: "the controller being reconciled is excluded even when not ready", + pods: []corev1.Pod{ + makePod("1", "true", false), + makePod("2", "true", true), + }, + currentBroker: "1", + expected: nil, + }, + { + testName: "not-ready broker-only node is ignored", + pods: []corev1.Pod{ + makePod("0", "false", false), + makePod("1", "true", true), + }, + currentBroker: "1", + expected: nil, + }, + { + testName: "combined node counts as a controller", + pods: []corev1.Pod{ + makePod("0", "true", false), + makePod("1", "true", true), + }, + currentBroker: "1", + expected: []string{"0"}, + }, + } + + for _, test := range tests { + t.Run(test.testName, func(t *testing.T) { + currentPod := &corev1.Pod{ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{v1beta1.BrokerIdLabelKey: test.currentBroker}}} + assert.Equal(t, test.expected, notReadyControllerBrokerIDs(test.pods, currentPod)) + }) + } +} + +func TestControllersBlockingRollingUpgrade(t *testing.T) { + pod := func(brokerID, isController string, ready bool) corev1.Pod { + readyStatus := corev1.ConditionFalse + if ready { + readyStatus = corev1.ConditionTrue + } + return corev1.Pod{ + ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{ + v1beta1.BrokerIdLabelKey: brokerID, + v1beta1.IsControllerNodeKey: isController, + }}, + Status: corev1.PodStatus{Conditions: []corev1.PodCondition{{Type: corev1.PodReady, Status: readyStatus}}}, + } + } + // broker 0: broker-only, broker 1: controller (not ready), broker 2: controller (ready) + pods := []corev1.Pod{pod("0", "false", true), pod("1", "true", false), pod("2", "true", true)} + currentPodFor := func(brokerID string) *corev1.Pod { + for i := range pods { + if pods[i].Labels[v1beta1.BrokerIdLabelKey] == brokerID { + return &pods[i] + } + } + return nil + } + + tests := []struct { + testName string + kRaftMode bool + currentBroker string + expected []string + }{ + { + testName: "not KRaft mode never gates", + kRaftMode: false, + currentBroker: "2", + expected: nil, + }, + { + testName: "rolling a broker-only node is not gated on controller readiness", + kRaftMode: true, + currentBroker: "0", + expected: nil, + }, + { + testName: "rolling a controller is blocked by another not-ready controller", + kRaftMode: true, + currentBroker: "2", + expected: []string{"1"}, + }, + { + testName: "rolling the not-ready controller itself is not blocked by its own readiness", + kRaftMode: true, + currentBroker: "1", + expected: nil, + }, + } + + for _, test := range tests { + t.Run(test.testName, func(t *testing.T) { + assert.Equal(t, test.expected, controllersBlockingRollingUpgrade(test.kRaftMode, pods, currentPodFor(test.currentBroker))) + }) + } +} diff --git a/pkg/resources/kafka/pod.go b/pkg/resources/kafka/pod.go index 7aeda6865..ee63004af 100644 --- a/pkg/resources/kafka/pod.go +++ b/pkg/resources/kafka/pod.go @@ -26,7 +26,6 @@ import ( corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" - "k8s.io/apimachinery/pkg/util/intstr" "github.com/banzaicloud/koperator/api/assets" apiutil "github.com/banzaicloud/koperator/api/util" @@ -106,24 +105,20 @@ fi`}, } if r.KafkaCluster.Spec.KRaftMode && brokerConfig.IsControllerNode() { - controllerlistenerPort, err := findControllerListenerPort(r.KafkaCluster) - if err != nil { - log.Error(err, "failed to find controller listener port") - } else { - kafkaContainer.ReadinessProbe = &corev1.Probe{ - ProbeHandler: corev1.ProbeHandler{ - TCPSocket: &corev1.TCPSocketAction{ - Port: intstr.IntOrString{ - Type: intstr.Int, - IntVal: controllerlistenerPort, - }, - }, + // Readiness reflects quorum membership: the controller pod is Ready only once it reports a + // leader/follower raft state. 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. The script runs in readiness mode (fail-closed while not yet in the quorum). + kafkaContainer.ReadinessProbe = &corev1.Probe{ + ProbeHandler: corev1.ProbeHandler{ + Exec: &corev1.ExecAction{ + Command: []string{"/bin/bash", "-c", "export KRAFT_HEALTH_CHECK_MODE=readiness\n" + assets.KraftControllerHealthcheckSh}, }, - InitialDelaySeconds: 0, - PeriodSeconds: 5, - TimeoutSeconds: 5, - FailureThreshold: 20, - } + }, + InitialDelaySeconds: 15, + PeriodSeconds: 15, + TimeoutSeconds: 10, + FailureThreshold: 6, } kafkaContainer.LivenessProbe = &corev1.Probe{ ProbeHandler: corev1.ProbeHandler{ @@ -132,8 +127,8 @@ fi`}, }, }, InitialDelaySeconds: 30, - PeriodSeconds: 10, - TimeoutSeconds: 5, + PeriodSeconds: 15, + TimeoutSeconds: 10, FailureThreshold: 6, } } @@ -641,14 +636,3 @@ func generateEnvConfig(brokerConfig *v1beta1.BrokerConfig, defaultEnvVars []core return mergedEnv } - -func findControllerListenerPort(kc *v1beta1.KafkaCluster) (int32, error) { - for _, listener := range kc.Spec.ListenersConfig.InternalListeners { - if listener.UsedForControllerCommunication { - return listener.ContainerPort, nil - } - } - - // If no controller listener is found, return an error - return 0, fmt.Errorf("no controller listener found") -} diff --git a/pkg/resources/kafka/service.go b/pkg/resources/kafka/service.go index 84e7e5c79..1a6da1ed3 100644 --- a/pkg/resources/kafka/service.go +++ b/pkg/resources/kafka/service.go @@ -55,10 +55,11 @@ func (r *Reconciler) service(id int32, _ *v1beta1.BrokerConfig) runtime.Object { r.KafkaCluster.Spec.ListenersConfig.GetServiceAnnotations(), r.KafkaCluster), Spec: corev1.ServiceSpec{ - Type: corev1.ServiceTypeClusterIP, - SessionAffinity: corev1.ServiceAffinityNone, - Selector: apiutil.MergeLabels(apiutil.LabelsForKafka(r.KafkaCluster.Name), map[string]string{v1beta1.BrokerIdLabelKey: fmt.Sprintf("%d", id)}), - Ports: usedPorts, + Type: corev1.ServiceTypeClusterIP, + SessionAffinity: corev1.ServiceAffinityNone, + Selector: apiutil.MergeLabels(apiutil.LabelsForKafka(r.KafkaCluster.Name), map[string]string{v1beta1.BrokerIdLabelKey: fmt.Sprintf("%d", id)}), + Ports: usedPorts, + PublishNotReadyAddresses: r.KafkaCluster.Spec.PublishNotReadyAddresses, }, } if r.KafkaCluster.Spec.LocalDebugEnabled { diff --git a/pkg/resources/kafka/service_test.go b/pkg/resources/kafka/service_test.go index f5a16448a..09c3e736d 100644 --- a/pkg/resources/kafka/service_test.go +++ b/pkg/resources/kafka/service_test.go @@ -32,182 +32,164 @@ import ( ) func TestService(t *testing.T) { + basePorts := []corev1.ServicePort{ + { + Name: "tcp-internal", + Protocol: "TCP", + Port: 29092, + TargetPort: intstr.FromInt(29092), + NodePort: 0, + }, + { + Name: "tcp-plaintext", + Protocol: "TCP", + Port: 29094, + TargetPort: intstr.FromInt(29094), + NodePort: 0, + }, + { + Name: "metrics", + Protocol: "TCP", + Port: 9020, + TargetPort: intstr.FromInt(9020), + NodePort: 0, + }, + } + + baseListeners := v1beta1.ListenersConfig{ + InternalListeners: []v1beta1.InternalListenerConfig{ + { + CommonListenerSpec: v1beta1.CommonListenerSpec{ + Name: "internal", + ContainerPort: 29092, + Type: "plaintext", + UsedForInnerBrokerCommunication: true, + }, + }, + }, + ExternalListeners: []v1beta1.ExternalListenerConfig{ + { + CommonListenerSpec: v1beta1.CommonListenerSpec{ + Name: "plaintext", + ContainerPort: 29094, + Type: "plaintext", + UsedForInnerBrokerCommunication: false, + }, + AccessMethod: corev1.ServiceTypeLoadBalancer, + }, + }, + } + + baseOwnerRef := []metav1.OwnerReference{ + { + APIVersion: "", + Kind: "", + Name: "kafka", + UID: "", + Controller: util.BoolPointer(true), + BlockOwnerDeletion: util.BoolPointer(true), + }, + } + testCases := []struct { testName string r *Reconciler expectedService *corev1.Service }{ { - testName: "Basic Internal And External Service", + testName: "ClusterIP service with publishNotReadyAddresses disabled (default)", r: &Reconciler{ Reconciler: resources.Reconciler{ KafkaCluster: &v1beta1.KafkaCluster{ - ObjectMeta: metav1.ObjectMeta{ - Name: "kafka", - Namespace: "kafka", - }, + ObjectMeta: metav1.ObjectMeta{Name: "kafka", Namespace: "kafka"}, Spec: v1beta1.KafkaClusterSpec{ - LocalDebugEnabled: false, - KRaftMode: false, - ListenersConfig: v1beta1.ListenersConfig{ - InternalListeners: []v1beta1.InternalListenerConfig{ - { - CommonListenerSpec: v1beta1.CommonListenerSpec{ - Name: "internal", - ContainerPort: 29092, - Type: "plaintext", - UsedForInnerBrokerCommunication: true, - }, - }, - }, - ExternalListeners: []v1beta1.ExternalListenerConfig{ - { - CommonListenerSpec: v1beta1.CommonListenerSpec{ - Name: "plaintext", - ContainerPort: 29094, - Type: "plaintext", - UsedForInnerBrokerCommunication: false, - }, - AccessMethod: corev1.ServiceTypeLoadBalancer, - }, - }, - }, + LocalDebugEnabled: false, + KRaftMode: false, + PublishNotReadyAddresses: false, + ListenersConfig: baseListeners, }, }, }, }, expectedService: &corev1.Service{ ObjectMeta: metav1.ObjectMeta{ - Name: "kafka-1", - Namespace: "kafka", - Labels: map[string]string{"app": "kafka", "brokerId": "1", "kafka_cr": "kafka"}, - Annotations: map[string]string{}, - OwnerReferences: []metav1.OwnerReference{ - { - APIVersion: "", - Kind: "", - Name: "kafka", - UID: "", - Controller: util.BoolPointer(true), - BlockOwnerDeletion: util.BoolPointer(true), - }, - }, + Name: "kafka-1", + Namespace: "kafka", + Labels: map[string]string{"app": "kafka", "brokerId": "1", "kafka_cr": "kafka"}, + Annotations: map[string]string{}, + OwnerReferences: baseOwnerRef, }, Spec: corev1.ServiceSpec{ - Type: corev1.ServiceTypeClusterIP, - SessionAffinity: corev1.ServiceAffinityNone, - Selector: apiutil.MergeLabels(apiutil.LabelsForKafka("kafka"), map[string]string{v1beta1.BrokerIdLabelKey: "1"}), - Ports: []corev1.ServicePort{ - { - Name: "tcp-internal", - Protocol: "TCP", - Port: 29092, - TargetPort: intstr.FromInt(29092), - NodePort: 0, - }, - { - Name: "tcp-plaintext", - Protocol: "TCP", - Port: 29094, - TargetPort: intstr.FromInt(29094), - NodePort: 0, - }, - { - Name: "metrics", - Protocol: "TCP", - Port: 9020, - TargetPort: intstr.FromInt(9020), - NodePort: 0, - }, - }, + Type: corev1.ServiceTypeClusterIP, + SessionAffinity: corev1.ServiceAffinityNone, + Selector: apiutil.MergeLabels(apiutil.LabelsForKafka("kafka"), map[string]string{v1beta1.BrokerIdLabelKey: "1"}), + Ports: basePorts, ClusterIP: "", PublishNotReadyAddresses: false, }, }, }, { - testName: "Basic Internal And External Service", + testName: "ClusterIP service with publishNotReadyAddresses enabled (URP sidecar mode)", r: &Reconciler{ Reconciler: resources.Reconciler{ KafkaCluster: &v1beta1.KafkaCluster{ - ObjectMeta: metav1.ObjectMeta{ - Name: "kafka", - Namespace: "kafka", - }, + ObjectMeta: metav1.ObjectMeta{Name: "kafka", Namespace: "kafka"}, Spec: v1beta1.KafkaClusterSpec{ - LocalDebugEnabled: true, - KRaftMode: false, - ListenersConfig: v1beta1.ListenersConfig{ - InternalListeners: []v1beta1.InternalListenerConfig{ - { - CommonListenerSpec: v1beta1.CommonListenerSpec{ - Name: "internal", - ContainerPort: 29092, - Type: "plaintext", - UsedForInnerBrokerCommunication: true, - }, - }, - }, - ExternalListeners: []v1beta1.ExternalListenerConfig{ - { - CommonListenerSpec: v1beta1.CommonListenerSpec{ - Name: "plaintext", - ContainerPort: 29094, - Type: "plaintext", - UsedForInnerBrokerCommunication: false, - }, - AccessMethod: corev1.ServiceTypeLoadBalancer, - }, - }, - }, + LocalDebugEnabled: false, + KRaftMode: false, + PublishNotReadyAddresses: true, + ListenersConfig: baseListeners, }, }, }, }, expectedService: &corev1.Service{ ObjectMeta: metav1.ObjectMeta{ - Name: "kafka-1", - Namespace: "kafka", - Labels: map[string]string{"app": "kafka", "brokerId": "1", "kafka_cr": "kafka"}, - Annotations: map[string]string{}, - OwnerReferences: []metav1.OwnerReference{ - { - APIVersion: "", - Kind: "", - Name: "kafka", - UID: "", - Controller: util.BoolPointer(true), - BlockOwnerDeletion: util.BoolPointer(true), - }, - }, + Name: "kafka-1", + Namespace: "kafka", + Labels: map[string]string{"app": "kafka", "brokerId": "1", "kafka_cr": "kafka"}, + Annotations: map[string]string{}, + OwnerReferences: baseOwnerRef, }, Spec: corev1.ServiceSpec{ - Type: corev1.ServiceTypeLoadBalancer, - SessionAffinity: corev1.ServiceAffinityNone, - Selector: apiutil.MergeLabels(apiutil.LabelsForKafka("kafka"), map[string]string{v1beta1.BrokerIdLabelKey: "1"}), - Ports: []corev1.ServicePort{ - { - Name: "tcp-internal", - Protocol: "TCP", - Port: 29092, - TargetPort: intstr.FromInt(29092), - NodePort: 0, - }, - { - Name: "tcp-plaintext", - Protocol: "TCP", - Port: 29094, - TargetPort: intstr.FromInt(29094), - NodePort: 0, - }, - { - Name: "metrics", - Protocol: "TCP", - Port: 9020, - TargetPort: intstr.FromInt(9020), - NodePort: 0, + Type: corev1.ServiceTypeClusterIP, + SessionAffinity: corev1.ServiceAffinityNone, + Selector: apiutil.MergeLabels(apiutil.LabelsForKafka("kafka"), map[string]string{v1beta1.BrokerIdLabelKey: "1"}), + Ports: basePorts, + ClusterIP: "", + PublishNotReadyAddresses: true, + }, + }, + }, + { + testName: "LoadBalancer service when localDebugEnabled", + r: &Reconciler{ + Reconciler: resources.Reconciler{ + KafkaCluster: &v1beta1.KafkaCluster{ + ObjectMeta: metav1.ObjectMeta{Name: "kafka", Namespace: "kafka"}, + Spec: v1beta1.KafkaClusterSpec{ + LocalDebugEnabled: true, + KRaftMode: false, + PublishNotReadyAddresses: false, + ListenersConfig: baseListeners, }, }, + }, + }, + expectedService: &corev1.Service{ + ObjectMeta: metav1.ObjectMeta{ + Name: "kafka-1", + Namespace: "kafka", + Labels: map[string]string{"app": "kafka", "brokerId": "1", "kafka_cr": "kafka"}, + Annotations: map[string]string{}, + OwnerReferences: baseOwnerRef, + }, + Spec: corev1.ServiceSpec{ + Type: corev1.ServiceTypeLoadBalancer, + SessionAffinity: corev1.ServiceAffinityNone, + Selector: apiutil.MergeLabels(apiutil.LabelsForKafka("kafka"), map[string]string{v1beta1.BrokerIdLabelKey: "1"}), + Ports: basePorts, ClusterIP: "", PublishNotReadyAddresses: false, }, diff --git a/pkg/resources/kafka/util.go b/pkg/resources/kafka/util.go index cfafbae14..1ecb0859b 100644 --- a/pkg/resources/kafka/util.go +++ b/pkg/resources/kafka/util.go @@ -67,9 +67,12 @@ func generateQuorumVoters(kafkaCluster *v1beta1.KafkaCluster, controllerListener return quorumVoters, nil } -// generateRandomClusterID() generates a based64-encoded random UUID with 16 bytes as the cluster ID -// it uses URL based64 encoding since that's what Kafka expects +// generateRandomClusterID() generates a base64-encoded random UUID with 16 bytes as the cluster ID. +// It uses URL base64 encoding without padding, matching Kafka's canonical org.apache.kafka.common.Uuid +// string form (Base64.getUrlEncoder().withoutPadding()) which yields 22 characters. The padded form +// (24 chars ending in "==") happens to be accepted by Kafka 3.9's Uuid.fromString (length <= 24), but it +// is non-canonical and rejected by newer Kafka versions, so we emit the padding-free canonical form. func generateRandomClusterID() string { randomUUID := uuid.New() - return base64.URLEncoding.EncodeToString(randomUUID[:]) + return base64.RawURLEncoding.EncodeToString(randomUUID[:]) } diff --git a/pkg/resources/kafka/util_test.go b/pkg/resources/kafka/util_test.go index d4c04045e..f5fe00078 100644 --- a/pkg/resources/kafka/util_test.go +++ b/pkg/resources/kafka/util_test.go @@ -29,11 +29,20 @@ func TestGenerateClusterID(t *testing.T) { test := make(map[string]bool, numOfIDs) for i := 0; i < numOfIDs; i++ { clusterID := generateRandomClusterID() - _, err := base64.URLEncoding.DecodeString(clusterID) + decoded, err := base64.RawURLEncoding.DecodeString(clusterID) if err != nil { t.Errorf("expected nil error, got: %v", err) } + // Kafka's Uuid is a 128-bit (16-byte) value; the base64 form must decode to exactly 16 bytes + // and use the canonical padding-free encoding (22 characters). + if len(decoded) != 16 { + t.Errorf("expected cluster ID to decode to 16 bytes, got %d bytes for %q", len(decoded), clusterID) + } + if len(clusterID) != 22 { + t.Errorf("expected canonical 22-character cluster ID, got %d characters for %q", len(clusterID), clusterID) + } + if test[clusterID] { t.Error("expected random cluster ID that does not collide with previous ones") } diff --git a/properties/pkg/utils.go b/properties/pkg/utils.go index 3cb08d3fb..3c8ca3f9d 100644 --- a/properties/pkg/utils.go +++ b/properties/pkg/utils.go @@ -45,8 +45,6 @@ func UnEscapeSeparators(s string) string { // Convert s string to slice of rune orig := []rune(s) - // Convert Separators to slice of rune - sep := []rune(Separators) // Create new slice holding the escaped string newSlice := make([]rune, 0) @@ -58,7 +56,7 @@ func UnEscapeSeparators(s string) string { // Set previous index by making sure that it's value is inbound prevIdx := Max(idx-1, 0) // Iterate over the separator characters. - for _, sp := range sep { + for _, sp := range Separators { // If there is a separator match and the previous is an escape character // then copy data up to the previous index to the new string then append // the current leaving out the previous character. @@ -90,8 +88,6 @@ func EscapeSeparators(s string) string { // Convert s string to slice of rune orig := []rune(s) - // Convert Separators to slice of rune - sep := []rune(Separators) // Create new slice holding the escaped string newSlice := make([]rune, 0) @@ -103,7 +99,7 @@ func EscapeSeparators(s string) string { // Set previous index by making sure that it's value is inbound prevIdx := Max(idx-1, 0) // Iterate over the separator characters - for _, sp := range sep { + for _, sp := range Separators { // If there is a separator match and the previous is not an escape character // then copy data up to the current index to the new string then append the // escape and the current characters. @@ -141,15 +137,13 @@ func GetSeparator(s string) (string, int, error) { // Convert s string to slice of rune r := []rune(s) - // Convert Separators to slice of rune - separators := []rune(Separators) // Iterate over the input string for idx, c := range r { // Avoid out of bound access prevIdx := Max(idx-1, 0) // Iterate over the list of separators - for _, sp := range separators { + for _, sp := range Separators { // If the current character is a separator and it is not escaped // than separator is found. if c == sp && r[prevIdx] != EscapeChar { diff --git a/renovate.json b/renovate.json index fcc01d149..dfbc6aa16 100644 --- a/renovate.json +++ b/renovate.json @@ -21,7 +21,8 @@ "managerFilePatterns": [ "/(^|/)Makefile$/", "/(^|/)Dockerfile$/", - "/(^|/)\\.github/workflows/regenerate-crd-docs\\.yaml$/" + "/(^|/)\\.github/workflows/regenerate-crd-docs\\.yaml$/", + "/(^|/)scripts/install_kustomize\\.sh$/" ], "matchStrings": [ "(?[A-Z][A-Z0-9_]*)_VERSION\\s*=\\s*(?[^\\s]+)\\s*#\\s*renovate:\\s*datasource=(?[^\\s]+)\\s*depName=(?[^\\s]+)(?:\\s*extractVersion=(?[^\\s]+))?" @@ -68,13 +69,55 @@ { "customType": "regex", "managerFilePatterns": [ - "/(^|/)\\.github/actions/kind-create/action\\.yaml$/" + "/(^|/)\\.github/actions/kind-create/action\\.yaml$/", + "/(^|/)\\.github/workflows/e2e-test\\.yaml$/" ], "matchStrings": [ "(?:default|version):\\s*'?(?[^\\s'#]+)'?\\s*#\\s*renovate:\\s*datasource=(?\\S+)\\s+depName=(?\\S+)" ], "datasourceTemplate": "{{datasource}}", "depNameTemplate": "{{packageName}}" + }, + { + "customType": "regex", + "description": "go install @ pins inside composite action shell steps (e.g. .github/actions/kind-create/action.yaml). Not covered by any built-in manager.", + "managerFilePatterns": [ + "/(^|/)\\.github/actions/kind-create/action\\.yaml$/" + ], + "matchStrings": [ + "go install \\S+@(?\\S+)\\s*#\\s*renovate:\\s*datasource=(?\\S+)\\s+depName=(?\\S+)" + ], + "datasourceTemplate": "{{datasource}}", + "depNameTemplate": "{{packageName}}" + }, + { + "customType": "regex", + "description": "Docker images embedded as flat 'repo:tag' strings in KafkaCluster sample manifests, benchmark infra YAML, and kustomize strategic-merge patches. The built-in helm-values manager only matches files literally named values.yaml/values.yml with a nested {repository,tag} shape, kustomize only reads the kustomization.yaml 'images:' transformer (unused in this repo), and the kubernetes manager has no default managerFilePatterns and is not configured here - so none of these were previously watched.", + "managerFilePatterns": [ + "/(^|/)config/samples/.*\\.yaml$/", + "/(^|/)docs/benchmarks/infrastructure/.*\\.yaml$/", + "/(^|/)tests/e2e/templates/.*\\.tmpl$/", + "/(^|/)config/base/manager/manager\\.yaml$/", + "/(^|/)config/overlays/.*/manager_auth_proxy_patch\\.yaml$/", + "/(^|/)hack/kafka-test-pod/manifest_certmanager\\.yaml$/" + ], + "matchStrings": [ + "[\"']?[a-zA-Z0-9-]+(?:\\.[a-zA-Z0-9-]+)*(?:/[a-zA-Z0-9._-]+)+:(?[a-zA-Z0-9._-]+)[\"']?\\s*#\\s*renovate:\\s*datasource=(?\\S+)\\s+depName=(?\\S+)" + ], + "datasourceTemplate": "{{datasource}}", + "depNameTemplate": "{{packageName}}" + }, + { + "customType": "regex", + "description": "Docker images embedded as nested helm-style {repository, tag} blocks in tests/e2e/templates/*.tmpl fixtures. The .tmpl extension is never matched by the built-in helm-values manager (values.yaml/.yml only), even though the shape matches.", + "managerFilePatterns": [ + "/(^|/)tests/e2e/templates/.*\\.tmpl$/" + ], + "matchStrings": [ + "tag:\\s*(?\\S+)\\s*#\\s*renovate:\\s*datasource=(?\\S+)\\s+depName=(?\\S+)" + ], + "datasourceTemplate": "{{datasource}}", + "depNameTemplate": "{{packageName}}" } ], "packageRules": [ @@ -95,13 +138,17 @@ { "groupName": "config samples docker images", "groupSlug": "config-samples-docker", + "description": "Was scoped to matchManagers: [helm-values], but that manager only reads files literally named values.yaml/values.yml with a nested {repository,tag} shape - it never matched anything under these paths, so this rule was a no-op. The images here are now tracked via explicit '# renovate:' comment annotations (custom.regex manager); see the customManagers entries covering config/samples, docs/benchmarks/infrastructure, tests/e2e/templates, config/base/manager, config/overlays, and hack/kafka-test-pod.", "matchFileNames": [ "config/samples/**", "docs/benchmarks/infrastructure/**", - "tests/e2e/templates/**" + "tests/e2e/templates/**", + "config/base/manager/**", + "config/overlays/**", + "hack/kafka-test-pod/**" ], "matchManagers": [ - "helm-values" + "custom.regex" ], "matchUpdateTypes": [ "major", @@ -120,7 +167,25 @@ "matchCurrentValue": "/^\\d+\\.\\d+\\.\\d+-ado?be-\\d+$/", "versioning": "regex:^(?\\d+)\\.(?\\d+)\\.(?\\d+)(?:-ado?be-(?\\d+))?$", "description": "Self-scoping: applies to ANY docker dep whose current pin is -adbe|adobe- (cruise-control uses 'adbe', zookeeper-operator 'adobe'; new adobe-fork images inherit this with no config change). Default docker versioning treats that suffix as an immutable compatibility string, so newer daily builds are never offered; capturing the date as a purely-numeric 'build' (handled like patch) makes e.g. 3.0.3-adbe-20250804 -> 3.0.3-adbe-20260722 and 0.2.15-adobe-20250923 track their newest build. matchCurrentValue requires the suffix, so only these deps are touched; the versioning's suffix is optional so the same image can still publish a plain-semver tag. Trailing '$' excludes '-rc'." - }, + }, + { + "matchDatasources": [ + "docker" + ], + "matchPackageNames": [ + "ghcr.io/adobe/zookeeper-operator/zookeeper" + ], + "allowedVersions": "/^\\d+\\.\\d+\\.\\d+-\\d+\\.\\d+\\.\\d+-adobe-\\d{8}$/", + "versioning": "loose" + }, + { + "matchPackageNames": [ + "ghcr.io/adobe/koperator/kafka" + ], + "matchCurrentValue": "/^\\d+\\.\\d+-\\d+\\.\\d+\\.\\d+(-jdk\\d+\\.\\d+\\.\\d+)?$/", + "versioning": "regex:^(?\\d+\\.\\d+)-(?\\d+)\\.(?\\d+)\\.(?\\d+)(?:-jdk(?\\d+)\\.\\d+\\.(?\\d+))?$", + "description": "Tag format is -[-jdk], e.g. 2.13-3.9.1 or 2.13-3.9.2-jdk21.0.11 (the latter shape shipped starting with the 3.9.2 release tag). Verified live against ghcr.io: default docker versioning parses currentVersion as just '2.13' and silently drops everything after the first hyphen, so Renovate can never see a real 3.9.1->3.9.2 bump (only digest-pin updates fire, via the docker pinDigests rule below) - this affects every '# renovate:' annotated occurrence of this image. Pinning the scala prefix as 'compatibility' (never auto-changed) and parsing the kafka release as major.minor.patch fixes real version comparison. The optional '-jdk..' suffix maps jdk major to 'build' and jdk update to 'revision' (Renovate's regex versioning caps out at 5 numeric levels: major/minor/patch/build/revision, each required to be purely numeric, so the always-0 jdk interim segment is matched but not captured); tags without the suffix default build/revision to 0 and still compare correctly against suffixed ones since a differing kafka patch alone already resolves ordering. This does NOT gate jdk-major jumps (e.g. jdk21.x -> jdk25.x) the way 'compatibility' gates scala - that's covered by the same operator-side migration as the apache/kafka 4.x bump below (see docker/kafka/Dockerfile's java_version comment); revisit if jdk25 lands before that migration." + }, { "groupName": "kafka docker images (major)", "groupSlug": "kafka-docker-major", diff --git a/run-local.sh b/run-local.sh index 80a53ce42..a022ede08 100755 --- a/run-local.sh +++ b/run-local.sh @@ -172,7 +172,7 @@ else fi ## Initialize Zookeeper and Kafka Cluster -kubectl apply -f config/samples/simplezookeeper.yaml -n zookeeper +kubectl apply -f config/samples/simpleZookeeper.yaml -n zookeeper if ! $LOCAL; then kubectl wait --for=condition=Ready pod -l app.kubernetes.io/name=kafka-operator -n kafka --timeout=120s diff --git a/scripts/install_kustomize.sh b/scripts/install_kustomize.sh index c4d25cb69..c01e166aa 100755 --- a/scripts/install_kustomize.sh +++ b/scripts/install_kustomize.sh @@ -14,11 +14,11 @@ # limitations under the License. -version=3.1.0 +KUSTOMIZE_VERSION=3.10.0 # renovate: datasource=github-releases depName=kubernetes-sigs/kustomize extractVersion=^kustomize/v(?.+)$ opsys=$(echo "$(uname -s)" | awk '{print tolower($0)}') # download the release -curl -O -L https://github.com/kubernetes-sigs/kustomize/releases/download/v${version}/kustomize_${version}_${opsys}_amd64 +curl -O -L https://github.com/kubernetes-sigs/kustomize/releases/download/v${KUSTOMIZE_VERSION}/kustomize_${KUSTOMIZE_VERSION}_${opsys}_amd64 # move to bin mkdir -p bin mv kustomize_*_${opsys}_amd64 bin/kustomize diff --git a/tests/e2e/const.go b/tests/e2e/const.go index 158cc8e64..f31d3e152 100644 --- a/tests/e2e/const.go +++ b/tests/e2e/const.go @@ -87,7 +87,8 @@ const ( contourName = "contour" // Kubernetes resource kinds. - podsResource = "pods" + podsResource = "pods" + configMapsResource = "configmaps" // Common string values and CLI flags/keys. falseString = "false" diff --git a/tests/e2e/k8s.go b/tests/e2e/k8s.go index 550e3df7a..77ef71a5f 100644 --- a/tests/e2e/k8s.go +++ b/tests/e2e/k8s.go @@ -483,6 +483,12 @@ func deleteK8sResourceNoErrNotFound(kubectlOptions k8s.KubectlOptions, timeout t return err } +func deleteK8sCRDNoErrNotFound(kubectlOptions k8s.KubectlOptions, crdName string) error { + clusterScopedOptions := kubectlOptions + clusterScopedOptions.Namespace = "" + return deleteK8sResourceNoErrNotFound(clusterScopedOptions, defaultDeletionTimeout, crdKind, crdName) +} + // applyK8sResourceManifestFromString applies the specified manifest in string format to the provided // kubectl context and namespace. func applyK8sResourceManifestFromString(kubectlOptions k8s.KubectlOptions, manifest string) error { diff --git a/tests/e2e/kafkacluster_brokers.go b/tests/e2e/kafkacluster_brokers.go new file mode 100644 index 000000000..db462c74c --- /dev/null +++ b/tests/e2e/kafkacluster_brokers.go @@ -0,0 +1,83 @@ +// Copyright 2026 Adobe. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//go:build e2e + +package e2e + +import ( + "encoding/json" + + "github.com/gruntwork-io/terratest/modules/k8s" + + "github.com/banzaicloud/koperator/api/v1beta1" +) + +// patchKafkaClusterBrokers fetches the running KafkaCluster CR, applies mutate to its current spec.brokers, +// and patches only that field back - leaving every other field untouched. The broker-removal test previously +// applied a second manifest to change broker count; besides the removal itself, that manifest can silently +// carry unrelated drift (e.g. simplekafkacluster.yaml's headlessServiceEnabled: false, which raced the +// operator's headless Service teardown against an in-flight broker removal). +// Patching only spec.brokers cannot introduce that kind of unrelated drift. +func patchKafkaClusterBrokers(kubectlOptions k8s.KubectlOptions, clusterName string, mutate func([]v1beta1.Broker) []v1beta1.Broker) error { + raw, err := runKubectlSilent(kubectlOptions, "get", "kafkacluster", clusterName, "-o", "json") + if err != nil { + return err + } + + var current struct { + Spec struct { + Brokers []v1beta1.Broker `json:"brokers"` + } `json:"spec"` + } + if err := json.Unmarshal([]byte(raw), ¤t); err != nil { + return err + } + + mergePatch, err := json.Marshal(map[string]any{ + "spec": map[string]any{ + "brokers": mutate(current.Spec.Brokers), + }, + }) + if err != nil { + return err + } + + _, err = runKubectlSilent(kubectlOptions, "patch", "kafkacluster", clusterName, "--type=merge", "-p", string(mergePatch)) + return err +} + +// removeKafkaClusterBrokers patches spec.brokers to exclude the given IDs. See patchKafkaClusterBrokers. +func removeKafkaClusterBrokers(kubectlOptions k8s.KubectlOptions, clusterName string, brokerIDsToRemove ...int32) error { + remove := make(map[int32]bool, len(brokerIDsToRemove)) + for _, id := range brokerIDsToRemove { + remove[id] = true + } + return patchKafkaClusterBrokers(kubectlOptions, clusterName, func(brokers []v1beta1.Broker) []v1beta1.Broker { + remaining := make([]v1beta1.Broker, 0, len(brokers)) + for _, broker := range brokers { + if !remove[broker.Id] { + remaining = append(remaining, broker) + } + } + return remaining + }) +} + +// addKafkaClusterBroker patches spec.brokers to append newBroker. See patchKafkaClusterBrokers. +func addKafkaClusterBroker(kubectlOptions k8s.KubectlOptions, clusterName string, newBroker v1beta1.Broker) error { + return patchKafkaClusterBrokers(kubectlOptions, clusterName, func(brokers []v1beta1.Broker) []v1beta1.Broker { + return append(brokers, newBroker) + }) +} diff --git a/tests/e2e/koperator_suite_test.go b/tests/e2e/koperator_suite_test.go index 189f41305..eae139116 100644 --- a/tests/e2e/koperator_suite_test.go +++ b/tests/e2e/koperator_suite_test.go @@ -18,6 +18,7 @@ package e2e import ( + "fmt" "testing" "github.com/gruntwork-io/terratest/modules/k8s" @@ -31,6 +32,21 @@ func TestKoperator(t *testing.T) { ginkgo.RunSpecs(t, "Koperator end to end test suite") } +// totalSpecsToRun and completedSpecCount back the "[k/N]" progress line printed after each spec below. +// The suite runs in series (no -p/--procs), so plain package vars are safe without synchronization. +var totalSpecsToRun int +var completedSpecCount int + +var _ = ginkgo.ReportBeforeSuite(func(report ginkgo.Report) { + totalSpecsToRun = report.PreRunStats.SpecsThatWillRun +}) + +var _ = ginkgo.ReportAfterEach(func(report ginkgo.SpecReport) { + completedSpecCount++ + fmt.Printf("--- [%d/%d specs done] %s: %s (took %s)\n", + completedSpecCount, totalSpecsToRun, report.FullText(), report.State, report.RunTime) +}) + var _ = ginkgo.BeforeSuite(func() { ginkgo.By("Acquiring K8s cluster") var kubeconfigPath string @@ -77,6 +93,7 @@ var _ = ginkgo.When("Testing e2e test altogether", ginkgo.Ordered, func() { testInstallKafkaCluster("../../config/samples/kraft/simplekafkacluster_kraft.yaml") testProduceConsumeInternal() testJmxExporter() + testKRaftBrokerScaling() testUninstallKafkaCluster() testUninstall() snapshotClusterAndCompare(snapshottedInfo) diff --git a/tests/e2e/templates/kcat.yaml.tmpl b/tests/e2e/templates/kcat.yaml.tmpl index e04978992..d90c25a0a 100644 --- a/tests/e2e/templates/kcat.yaml.tmpl +++ b/tests/e2e/templates/kcat.yaml.tmpl @@ -7,7 +7,7 @@ spec: serviceAccount: {{or .ServiceAccount "default"}} containers: - name: kafka-client - image: edenhill/kcat:1.7.0 + image: edenhill/kcat:1.7.1 # renovate: datasource=docker depName=edenhill/kcat # Just spin & wait forever command: [ "/bin/sh", "-c", "--" ] args: [ "while true; do sleep 3000; done;" ] diff --git a/tests/e2e/templates/zookeeper_cluster.yaml.tmpl b/tests/e2e/templates/zookeeper_cluster.yaml.tmpl index be89f92fc..aebf9f4c0 100644 --- a/tests/e2e/templates/zookeeper_cluster.yaml.tmpl +++ b/tests/e2e/templates/zookeeper_cluster.yaml.tmpl @@ -7,6 +7,6 @@ spec: replicas: {{ or .Replicas 1 }} image: repository: ghcr.io/adobe/zookeeper-operator/zookeeper - tag: 3.8.4-0.2.15-adobe-20250923 + tag: 3.8.4-0.2.15-adobe-20260818 # renovate: datasource=docker depName=ghcr.io/adobe/zookeeper-operator/zookeeper persistence: reclaimPolicy: Delete diff --git a/tests/e2e/test_broker_removal.go b/tests/e2e/test_broker_removal.go index 9ca5ff8df..2c1d074f5 100644 --- a/tests/e2e/test_broker_removal.go +++ b/tests/e2e/test_broker_removal.go @@ -74,9 +74,10 @@ func testBatchedBrokerRemoval() bool { }, batchedBrokerRemovalTimeout, batchedBrokerRemovalPollInterval).Should(gomega.BeTrue()) }) - ginkgo.It("Applying 3-broker manifest to trigger removal of brokers 3 and 4", func() { - ginkgo.By("Patching KafkaCluster to remove brokers 3 and 4") - applyK8sResourceManifest(kubectlOptions, "../../config/samples/simplekafkacluster.yaml") + ginkgo.It("Removing brokers 3 and 4 from the running KafkaCluster", func() { + ginkgo.By("Fetching the running KafkaCluster and patching out brokers 3 and 4") + err := removeKafkaClusterBrokers(kubectlOptions, kafkaClusterName, 3, 4) + gomega.Expect(err).NotTo(gomega.HaveOccurred()) }) ginkgo.It("Waiting for exactly one remove_broker CruiseControlOperation to be created", func() { diff --git a/tests/e2e/test_kraft_broker_scaling.go b/tests/e2e/test_kraft_broker_scaling.go new file mode 100644 index 000000000..0a19c4e9b --- /dev/null +++ b/tests/e2e/test_kraft_broker_scaling.go @@ -0,0 +1,302 @@ +// Copyright 2026 Adobe. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//go:build e2e + +package e2e + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "sort" + "strings" + + "github.com/gruntwork-io/terratest/modules/k8s" + ginkgo "github.com/onsi/ginkgo/v2" + gomega "github.com/onsi/gomega" + + "github.com/banzaicloud/koperator/api/v1alpha1" + "github.com/banzaicloud/koperator/api/v1beta1" +) + +// testKRaftBrokerScaling upscales a KRaft cluster from 3 broker-only nodes to 4 (add broker 103) and then +// downscales it back to 3, asserting Cruise Control drives each direction with exactly one +// add_broker / remove_broker operation, the broker-only node set tracks the spec by exact id, and the 3 +// controller-only nodes (ids 0,1,2) are never touched (controller-only nodes are not CC brokers). The +// assertions check exact broker/controller ids - not just pod counts - so a manifest that silently swaps +// ids (rather than adding a single broker) would fail instead of passing green. +// +// This is the regression test for #301, and it deliberately exercises BOTH directions: +// - UPSCALE is the case a "stop hashing capacity.json" fix would have silently broken: the capacity roll +// is what loads the new broker's exact capacity into CC, so the fix must keep rolling on add and +// sequence the add_broker op after the roll settles. +// - DOWNSCALE is the case that stalled on master: the pre-downscale capacity roll kept CC un-ready so the +// remove_broker op was never created. +func testKRaftBrokerScaling() bool { + return ginkgo.When("KRaft broker scaling: upscale then downscale, controllers untouched", func() { + var kubectlOptions k8s.KubectlOptions + var err error + + ginkgo.It("Acquiring K8s config and context", func() { + kubectlOptions, err = kubectlOptionsForCurrentContext() + gomega.Expect(err).NotTo(gomega.HaveOccurred()) + kubectlOptions.Namespace = koperatorLocalHelmDescriptor.Namespace + }) + + ginkgo.It("Waiting for Cruise Control to be ready and settled before upscale", func() { + ginkgo.By("Ensuring the KafkaCluster is running") + gomega.Expect(waitForKafkaClusterWithPodStatusCheck(kubectlOptions, kafkaClusterName, kafkaClusterResourceReadinessTimeout)).NotTo(gomega.HaveOccurred()) + ginkgo.By("Waiting until no Cruise Control operation is in flight (initial rebalance finished)") + gomega.Eventually(context.Background(), func() (bool, error) { + return hasNoInFlightCruiseControlOperation(kubectlOptions) + }, batchedBrokerRemovalTimeout, batchedBrokerRemovalPollInterval).Should(gomega.BeTrue()) + ginkgo.By("Waiting until the Cruise Control Deployment is fully rolled out (single Ready replica)") + gomega.Eventually(context.Background(), func() (bool, error) { + return isCruiseControlDeploymentRolledOut(kubectlOptions) + }, batchedBrokerRemovalTimeout, batchedBrokerRemovalPollInterval).Should(gomega.BeTrue()) + }) + + ginkgo.It("Asserting the cluster starts with broker-only nodes 100,101,102 and controllers 0,1,2", func() { + ids, err := brokerPodIDs(kubectlOptions) + gomega.Expect(err).NotTo(gomega.HaveOccurred()) + gomega.Expect(ids).To(gomega.ConsistOf("100", "101", "102"), "unexpected broker-only ids before scaling") + ids, err = controllerPodIDs(kubectlOptions) + gomega.Expect(err).NotTo(gomega.HaveOccurred()) + gomega.Expect(ids).To(gomega.ConsistOf("0", "1", "2"), "unexpected controller-only ids before scaling") + }) + + // --- Upscale: add broker 103 --- + + ginkgo.It("Adding broker 103 to the running KafkaCluster", func() { + ginkgo.By("Fetching the running KafkaCluster and adding broker 103") + // Matches the "broker" config group simplekafkacluster_kraft.yaml assigns brokers 100-102. + err := addKafkaClusterBroker(kubectlOptions, kafkaClusterName, v1beta1.Broker{Id: 103, BrokerConfigGroup: "broker"}) + gomega.Expect(err).NotTo(gomega.HaveOccurred()) + }) + + ginkgo.It("Waiting for exactly one add_broker CruiseControlOperation", func() { + gomega.Eventually(context.Background(), func() (bool, error) { + return hasExactlyNBrokerOperations(kubectlOptions, v1alpha1.OperationAddBroker, 1) + }, batchedBrokerRemovalTimeout, batchedBrokerRemovalPollInterval).Should(gomega.BeTrue()) + }) + + ginkgo.It("Waiting for broker 103 to join (broker-only nodes 100,101,102,103) and the cluster to be healthy", func() { + gomega.Eventually(context.Background(), func() ([]string, error) { + return brokerPodIDs(kubectlOptions) + }, batchedBrokerRemovalTimeout, batchedBrokerRemovalPollInterval).Should(gomega.ConsistOf("100", "101", "102", "103")) + gomega.Expect(waitForKafkaClusterWithPodStatusCheck(kubectlOptions, kafkaClusterName, kafkaClusterResourceReadinessTimeout)).NotTo(gomega.HaveOccurred()) + }) + + ginkgo.It("Asserting controllers are untouched after upscale (still exactly controllers 0,1,2)", func() { + ids, err := controllerPodIDs(kubectlOptions) + gomega.Expect(err).NotTo(gomega.HaveOccurred()) + gomega.Expect(ids).To(gomega.ConsistOf("0", "1", "2"), "controller-only nodes must not be affected by an upscale") + }) + + ginkgo.It("Asserting Cruise Control's capacity.json gained a real entry for the added broker 103", func() { + // add_broker runs with AllowCapacityEstimation=true, so a healthy add alone does not prove the + // operator wrote broker 103's capacity - CC could have estimated it. Assert the per-broker entry + // is actually present so a regression that stops generating/rolling capacity.json is caught. + gomega.Eventually(context.Background(), func() ([]string, error) { + return cruiseControlCapacityBrokerIDs(kubectlOptions) + }, batchedBrokerRemovalTimeout, batchedBrokerRemovalPollInterval).Should(gomega.ContainElement("103"), + "broker 103's capacity must be written to capacity.json, not left to CC estimation") + }) + + ginkgo.It("Asserting the running CC pod's template carries the current capacity.json hash", func() { + // The per-broker entry existing in the ConfigMap is not enough: the capacity must also have been + // hashed into the pod template and rolled out. Assert the hash on the *running* CC pod (not just + // the Deployment template) matches the current capacity.json - so a Ready CC pod started from that + // capacity. It does not by itself prove the CC process re-read the file, but combined with a Ready + // pod it is strong evidence the roll landed rather than add_broker relying on capacity estimation. + gomega.Eventually(context.Background(), func() (bool, error) { + return runningCruiseControlPodHasCurrentCapacityHash(kubectlOptions) + }, batchedBrokerRemovalTimeout, batchedBrokerRemovalPollInterval).Should(gomega.BeTrue(), + "the running CC pod's template must carry the hash of the current capacity.json") + }) + + // --- Downscale: remove broker 103 --- + + ginkgo.It("Waiting for Cruise Control to be settled again before downscale", func() { + gomega.Eventually(context.Background(), func() (bool, error) { + return hasNoInFlightCruiseControlOperation(kubectlOptions) + }, batchedBrokerRemovalTimeout, batchedBrokerRemovalPollInterval).Should(gomega.BeTrue()) + gomega.Eventually(context.Background(), func() (bool, error) { + return isCruiseControlDeploymentRolledOut(kubectlOptions) + }, batchedBrokerRemovalTimeout, batchedBrokerRemovalPollInterval).Should(gomega.BeTrue()) + }) + + ginkgo.It("Removing broker 103 from the running KafkaCluster", func() { + ginkgo.By("Fetching the running KafkaCluster and patching out broker 103") + err := removeKafkaClusterBrokers(kubectlOptions, kafkaClusterName, 103) + gomega.Expect(err).NotTo(gomega.HaveOccurred()) + }) + + ginkgo.It("Waiting for exactly one remove_broker CruiseControlOperation", func() { + gomega.Eventually(context.Background(), func() (bool, error) { + return hasExactlyNBrokerOperations(kubectlOptions, v1alpha1.OperationRemoveBroker, 1) + }, batchedBrokerRemovalTimeout, batchedBrokerRemovalPollInterval).Should(gomega.BeTrue()) + }) + + ginkgo.It("Waiting for broker 103 to be removed (broker-only nodes back to 100,101,102) and the cluster to be healthy", func() { + gomega.Eventually(context.Background(), func() ([]string, error) { + return brokerPodIDs(kubectlOptions) + }, batchedBrokerRemovalTimeout, batchedBrokerRemovalPollInterval).Should(gomega.ConsistOf("100", "101", "102")) + gomega.Expect(waitForKafkaClusterWithPodStatusCheck(kubectlOptions, kafkaClusterName, kafkaClusterResourceReadinessTimeout)).NotTo(gomega.HaveOccurred()) + }) + + ginkgo.It("Asserting controllers are untouched after downscale (still exactly controllers 0,1,2)", func() { + ids, err := controllerPodIDs(kubectlOptions) + gomega.Expect(err).NotTo(gomega.HaveOccurred()) + gomega.Expect(ids).To(gomega.ConsistOf("0", "1", "2"), "controller-only nodes must not be affected by a downscale") + }) + + ginkgo.It("Asserting Cruise Control's capacity.json dropped the removed broker 103", func() { + // Once broker 103's pod is gone and it leaves the status, the merge stops preserving its entry + // and capacity.json shrinks back - the departing broker's capacity must not linger indefinitely. + gomega.Eventually(context.Background(), func() ([]string, error) { + return cruiseControlCapacityBrokerIDs(kubectlOptions) + }, batchedBrokerRemovalTimeout, batchedBrokerRemovalPollInterval).ShouldNot(gomega.ContainElement("103"), + "broker 103's capacity must be removed from capacity.json once it is fully downscaled") + }) + }) +} + +// hasExactlyNBrokerOperations returns true when exactly n CruiseControlOperations whose current task is the +// given operation type exist in the namespace. +func hasExactlyNBrokerOperations(kubectlOptions k8s.KubectlOptions, operation v1alpha1.CruiseControlTaskOperation, n int) (bool, error) { + ops, err := getK8sResources(kubectlOptions, + []string{"cruisecontroloperation"}, + "", + "", + "-o", "jsonpath={range .items[*]}{.status.currentTask.operation}{'\\n'}{end}", + ) + if err != nil { + return false, err + } + count := 0 + for _, op := range ops { + if op == string(operation) { + count++ + } + } + return count == n, nil +} + +// brokerPodIDs returns the sorted broker ids (brokerId label) of the Running broker-only pods +// (isControllerNode=false) in the namespace. +func brokerPodIDs(kubectlOptions k8s.KubectlOptions) ([]string, error) { + return podBrokerIDs(kubectlOptions, kafkaLabelSelectorBrokers) +} + +// controllerPodIDs returns the sorted broker ids (brokerId label) of the Running controller-only pods +// (isControllerNode=true) in the namespace. +func controllerPodIDs(kubectlOptions k8s.KubectlOptions) ([]string, error) { + return podBrokerIDs(kubectlOptions, kafkaLabelSelectorControllers) +} + +// podBrokerIDs returns the sorted brokerId label values of the Running pods matching roleSelector. Asserting +// on the exact id set (rather than just a pod count) is what lets the scaling test catch a manifest that +// swaps ids instead of adding/removing a single node. +func podBrokerIDs(kubectlOptions k8s.KubectlOptions, roleSelector string) ([]string, error) { + ids, err := getK8sResources(kubectlOptions, + []string{podsResource}, + v1beta1.KafkaCRLabelKey+"="+kafkaClusterName+","+roleSelector, + "", + "--field-selector=status.phase=Running", + "-o", "jsonpath={range .items[*]}{.metadata.labels.brokerId}{'\\n'}{end}", + ) + if err != nil { + return nil, err + } + sort.Strings(ids) + return ids, nil +} + +// cruiseControlCapacityBrokerIDs returns the sorted broker ids present in Cruise Control's capacity.json +// (read from the CC ConfigMap). Asserting a scaled broker appears/disappears here proves the operator +// actually generated and rolled its per-broker capacity, independently of add_broker succeeding via Cruise +// Control's capacity estimation (AllowCapacityEstimation) - so a regression that stops writing per-broker +// capacity is caught rather than masked by estimation. +func cruiseControlCapacityBrokerIDs(kubectlOptions k8s.KubectlOptions) ([]string, error) { + lines, err := getK8sResources(kubectlOptions, + []string{configMapsResource}, + v1beta1.KafkaCRLabelKey+"="+kafkaClusterName+",app=cruisecontrol", + "", + "-o", "jsonpath={range .items[*]}{.data.capacity\\.json}{end}", + ) + if err != nil { + return nil, err + } + raw := strings.TrimSpace(strings.Join(lines, "\n")) + if raw == "" { + return nil, nil + } + var parsed struct { + BrokerCapacities []struct { + BrokerID string `json:"brokerId"` + } `json:"brokerCapacities"` + } + if err := json.Unmarshal([]byte(raw), &parsed); err != nil { + return nil, err + } + ids := make([]string, 0, len(parsed.BrokerCapacities)) + for _, bc := range parsed.BrokerCapacities { + ids = append(ids, bc.BrokerID) + } + sort.Strings(ids) + return ids, nil +} + +// runningCruiseControlPodHasCurrentCapacityHash reports whether the Running CC pod's template carries the hash +// of the CC ConfigMap's current capacity.json (the annotation koperator stamps in GeneratePodAnnotations). +// Reading the pod (not the Deployment template) shows what the currently-running CC actually started from, so +// with a Ready pod it is evidence the capacity roll landed rather than add_broker relying on capacity +// estimation. The hash is computed the same way the operator does (hex(sha256(capacity.json))). During a +// rollout there may briefly be two CC pods with different hashes; requiring an exact single-value match makes +// this true only once the roll has settled to the current-capacity pod. +func runningCruiseControlPodHasCurrentCapacityHash(kubectlOptions k8s.KubectlOptions) (bool, error) { + ccSelector := v1beta1.KafkaCRLabelKey + "=" + kafkaClusterName + ",app=cruisecontrol" + + cmLines, err := getK8sResources(kubectlOptions, + []string{configMapsResource}, + ccSelector, + "", + "-o", "jsonpath={range .items[*]}{.data.capacity\\.json}{end}", + ) + if err != nil { + return false, err + } + // Do not trim the capacity.json - it is hashed byte-for-byte by the operator. + capacityJSON := strings.Join(cmLines, "\n") + if capacityJSON == "" { + return false, nil + } + + podLines, err := getK8sResources(kubectlOptions, + []string{podsResource}, + ccSelector, + "", + "--field-selector=status.phase=Running", + "-o", "jsonpath={range .items[*]}{.metadata.annotations.cruiseControlCapacity\\.json}{'\\n'}{end}", + ) + if err != nil { + return false, err + } + runningHash := strings.TrimSpace(strings.Join(podLines, "\n")) + sum := sha256.Sum256([]byte(capacityJSON)) + return runningHash != "" && runningHash == hex.EncodeToString(sum[:]), nil +} diff --git a/tests/e2e/uninstall.go b/tests/e2e/uninstall.go index 5ab3f6f06..f887e1568 100644 --- a/tests/e2e/uninstall.go +++ b/tests/e2e/uninstall.go @@ -16,6 +16,7 @@ package e2e import ( + "context" "fmt" "strings" @@ -41,21 +42,7 @@ func requireUninstallingKoperatorHelmChart(kubectlOptions k8s.KubectlOptions) { gomega.Expect(err).NotTo(gomega.HaveOccurred()) ginkgo.By("Verifying Koperator helm chart resources cleanup") - k8sResourceKinds, err := listK8sResourceKinds(kubectlOptions, "") - gomega.Expect(err).ShouldNot(gomega.HaveOccurred()) - - koperatorAvailableResourceKinds := stringSlicesInstersect(koperatorCRDs(), k8sResourceKinds) - koperatorAvailableResourceKinds = append(koperatorAvailableResourceKinds, basicK8sResourceKinds()...) - - remainedResources, err := getK8sResources(kubectlOptions, - koperatorAvailableResourceKinds, - fmt.Sprintf(managedByHelmLabelTemplate, koperatorLocalHelmDescriptor.ReleaseName), - "", - kubectlArgGoTemplateKindNameNamespace, - "--all-namespaces") - - gomega.Expect(err).ShouldNot(gomega.HaveOccurred()) - gomega.Expect(remainedResources).Should(gomega.BeEmpty()) + requireHelmManagedResourcesRemoved(kubectlOptions, koperatorCRDs(), koperatorLocalHelmDescriptor.ReleaseName) }) } @@ -63,7 +50,7 @@ func requireUninstallingKoperatorHelmChart(kubectlOptions k8s.KubectlOptions) { func requireRemoveKoperatorCRDs(kubectlOptions k8s.KubectlOptions) { ginkgo.It("Removing koperator CRDs", func() { for _, crd := range koperatorCRDs() { - err := deleteK8sResourceNoErrNotFound(kubectlOptions, defaultDeletionTimeout, crdKind, crd) + err := deleteK8sCRDNoErrNotFound(kubectlOptions, crd) gomega.Expect(err).ShouldNot(gomega.HaveOccurred()) } }) @@ -87,21 +74,7 @@ func requireUninstallingZookeeperOperatorHelmChart(kubectlOptions k8s.KubectlOpt gomega.Expect(err).NotTo(gomega.HaveOccurred()) ginkgo.By("Verifying Zookeeper-operator helm chart resources cleanup") - k8sResourceKinds, err := listK8sResourceKinds(kubectlOptions, "") - gomega.Expect(err).ShouldNot(gomega.HaveOccurred()) - - zookeeperAvailableResourceKinds := stringSlicesInstersect(dependencyCRDs.Zookeeper(), k8sResourceKinds) - zookeeperAvailableResourceKinds = append(zookeeperAvailableResourceKinds, basicK8sResourceKinds()...) - - remainedResources, err := getK8sResources(kubectlOptions, - zookeeperAvailableResourceKinds, - fmt.Sprintf(managedByHelmLabelTemplate, zookeeperOperatorHelmDescriptor.ReleaseName), - "", - kubectlArgGoTemplateKindNameNamespace, - "--all-namespaces") - gomega.Expect(err).ShouldNot(gomega.HaveOccurred()) - - gomega.Expect(remainedResources).Should(gomega.BeEmpty()) + requireHelmManagedResourcesRemoved(kubectlOptions, dependencyCRDs.Zookeeper(), zookeeperOperatorHelmDescriptor.ReleaseName) }) } @@ -109,7 +82,7 @@ func requireUninstallingZookeeperOperatorHelmChart(kubectlOptions k8s.KubectlOpt func requireRemoveZookeeperOperatorCRDs(kubectlOptions k8s.KubectlOptions) { ginkgo.It("Removing zookeeper-operator CRDs", func() { for _, crd := range dependencyCRDs.Zookeeper() { - err := deleteK8sResourceNoErrNotFound(kubectlOptions, defaultDeletionTimeout, crdKind, crd) + err := deleteK8sCRDNoErrNotFound(kubectlOptions, crd) gomega.Expect(err).ShouldNot(gomega.HaveOccurred()) } }) @@ -134,21 +107,7 @@ func requireUninstallingPrometheusOperatorHelmChart(kubectlOptions k8s.KubectlOp ginkgo.By("Verifying Prometheus-operator helm chart resources cleanup") - k8sResourceKinds, err := listK8sResourceKinds(kubectlOptions, "") - gomega.Expect(err).ShouldNot(gomega.HaveOccurred()) - - prometheusAvailableResourceKinds := stringSlicesInstersect(dependencyCRDs.Prometheus(), k8sResourceKinds) - prometheusAvailableResourceKinds = append(prometheusAvailableResourceKinds, basicK8sResourceKinds()...) - - remainedResources, err := getK8sResources(kubectlOptions, - prometheusAvailableResourceKinds, - fmt.Sprintf(managedByHelmLabelTemplate, prometheusOperatorHelmDescriptor.ReleaseName), - "", - kubectlArgGoTemplateKindNameNamespace, - "--all-namespaces") - gomega.Expect(err).ShouldNot(gomega.HaveOccurred()) - - gomega.Expect(remainedResources).Should(gomega.BeEmpty()) + requireHelmManagedResourcesRemoved(kubectlOptions, dependencyCRDs.Prometheus(), prometheusOperatorHelmDescriptor.ReleaseName) }) } @@ -156,7 +115,7 @@ func requireUninstallingPrometheusOperatorHelmChart(kubectlOptions k8s.KubectlOp func requireRemovePrometheusOperatorCRDs(kubectlOptions k8s.KubectlOptions) { ginkgo.It("Removing prometheus-operator CRDs", func() { for _, crd := range dependencyCRDs.Prometheus() { - err := deleteK8sResourceNoErrNotFound(kubectlOptions, defaultDeletionTimeout, crdKind, crd) + err := deleteK8sCRDNoErrNotFound(kubectlOptions, crd) gomega.Expect(err).ShouldNot(gomega.HaveOccurred()) } }) @@ -181,21 +140,7 @@ func requireUninstallingCertManagerHelmChart(kubectlOptions k8s.KubectlOptions) ginkgo.By("Verifying Cert-manager helm chart resources cleanup") - k8sResourceKinds, err := listK8sResourceKinds(kubectlOptions, "") - gomega.Expect(err).ShouldNot(gomega.HaveOccurred()) - - certManagerAvailableResourceKinds := stringSlicesInstersect(dependencyCRDs.CertManager(), k8sResourceKinds) - certManagerAvailableResourceKinds = append(certManagerAvailableResourceKinds, basicK8sResourceKinds()...) - - remainedResources, err := getK8sResources(kubectlOptions, - certManagerAvailableResourceKinds, - fmt.Sprintf(managedByHelmLabelTemplate, certManagerHelmDescriptor.ReleaseName), - "", - kubectlArgGoTemplateKindNameNamespace, - "--all-namespaces") - gomega.Expect(err).ShouldNot(gomega.HaveOccurred()) - - gomega.Expect(remainedResources).Should(gomega.BeEmpty()) + requireHelmManagedResourcesRemoved(kubectlOptions, dependencyCRDs.CertManager(), certManagerHelmDescriptor.ReleaseName) }) } @@ -204,7 +149,7 @@ func requireRemoveCertManagerCRDs(kubectlOptions k8s.KubectlOptions) { ginkgo.It("Removing cert-manager CRDs", func() { // First, try to remove CRDs detected by the dependencyCRDs system for _, crd := range dependencyCRDs.CertManager() { - err := deleteK8sResourceNoErrNotFound(kubectlOptions, defaultDeletionTimeout, crdKind, crd) + err := deleteK8sCRDNoErrNotFound(kubectlOptions, crd) gomega.Expect(err).ShouldNot(gomega.HaveOccurred()) } @@ -219,7 +164,7 @@ func requireRemoveCertManagerCRDs(kubectlOptions k8s.KubectlOptions) { } for _, crd := range knownCertManagerCRDs { - err := deleteK8sResourceNoErrNotFound(kubectlOptions, defaultDeletionTimeout, crdKind, crd) + err := deleteK8sCRDNoErrNotFound(kubectlOptions, crd) if err != nil && !isKubectlNotFoundError(err) { ginkgo.By(fmt.Sprintf("Warning: Failed to delete CRD %s: %v", crd, err)) } @@ -255,28 +200,34 @@ func requireUninstallingContourHelmChart(kubectlOptions k8s.KubectlOptions) { ginkgo.By("Verifying Project Contour helm chart resources cleanup") - k8sResourceKinds, err := listK8sResourceKinds(kubectlOptions, "") - gomega.Expect(err).ShouldNot(gomega.HaveOccurred()) + requireHelmManagedResourcesRemoved(kubectlOptions, dependencyCRDs.Contour(), contourIngressControllerHelmDescriptor.ReleaseName) + }) +} - contourAvailableResourceKinds := stringSlicesInstersect(dependencyCRDs.Contour(), k8sResourceKinds) - contourAvailableResourceKinds = append(contourAvailableResourceKinds, basicK8sResourceKinds()...) +func requireHelmManagedResourcesRemoved(kubectlOptions k8s.KubectlOptions, crdResourceKinds []string, releaseName string) { + k8sResourceKinds, err := listK8sResourceKinds(kubectlOptions, "") + gomega.Expect(err).ShouldNot(gomega.HaveOccurred()) - remainedResources, err := getK8sResources(kubectlOptions, - contourAvailableResourceKinds, - fmt.Sprintf(managedByHelmLabelTemplate, contourIngressControllerHelmDescriptor.ReleaseName), + availableResourceKinds := stringSlicesInstersect(crdResourceKinds, k8sResourceKinds) + availableResourceKinds = append(availableResourceKinds, basicK8sResourceKinds()...) + selector := fmt.Sprintf(managedByHelmLabelTemplate, releaseName) + + // Helm uninstall can return before Kubernetes stops serving terminating objects. + gomega.Eventually(context.Background(), func() ([]string, error) { + return getK8sResourcesQuiet(kubectlOptions, + availableResourceKinds, + selector, "", kubectlArgGoTemplateKindNameNamespace, "--all-namespaces") - gomega.Expect(err).ShouldNot(gomega.HaveOccurred()) - - gomega.Expect(remainedResources).Should(gomega.BeEmpty()) - }) + }, defaultDeletionTimeout, waitResourceConditionRetryInterval).Should(gomega.BeEmpty(), + "Helm-managed resources for release %q remained after uninstall", releaseName) } func requireRemoveContourCRDs(kubectlOptions k8s.KubectlOptions) { ginkgo.It("Removing Contour Ingress Controller CRDs", func() { for _, crd := range dependencyCRDs.Contour() { - err := deleteK8sResourceNoErrNotFound(kubectlOptions, defaultDeletionTimeout, crdKind, crd) + err := deleteK8sCRDNoErrNotFound(kubectlOptions, crd) gomega.Expect(err).ShouldNot(gomega.HaveOccurred()) } }) diff --git a/tests/e2e/versions.go b/tests/e2e/versions.go index 95cb76511..67605d12b 100644 --- a/tests/e2e/versions.go +++ b/tests/e2e/versions.go @@ -22,11 +22,11 @@ const ( CertManagerVersion = "v1.21.1" // renovate: datasource=helm depName=cert-manager registryUrl=https://charts.jetstack.io // ContourVersion is the version of Contour ingress controller Helm chart - ContourVersion = "0.6.0" // renovate: datasource=helm depName=contour registryUrl=https://projectcontour.github.io/helm-charts + ContourVersion = "0.7.0" // renovate: datasource=helm depName=contour registryUrl=https://projectcontour.github.io/helm-charts // PrometheusOperatorVersion is the version of kube-prometheus-stack Helm chart - PrometheusOperatorVersion = "88.2.0" // renovate: datasource=helm depName=kube-prometheus-stack registryUrl=https://prometheus-community.github.io/helm-charts + PrometheusOperatorVersion = "88.5.0" // renovate: datasource=helm depName=kube-prometheus-stack registryUrl=https://prometheus-community.github.io/helm-charts // ZookeeperOperatorVersion is the version of zookeeper-operator - ZookeeperOperatorVersion = "0.2.15-adobe-20260423" // renovate: datasource=docker depName=ghcr.io/adobe/helm-charts/zookeeper-operator + ZookeeperOperatorVersion = "0.2.15-adobe-20260818" // renovate: datasource=docker depName=ghcr.io/adobe/helm-charts/zookeeper-operator ) diff --git a/third_party/github.com/banzaicloud/go-cruise-control/deploy/docker-compose.yml b/third_party/github.com/banzaicloud/go-cruise-control/deploy/docker-compose.yml index 22d034ff8..20b7d2d0d 100644 --- a/third_party/github.com/banzaicloud/go-cruise-control/deploy/docker-compose.yml +++ b/third_party/github.com/banzaicloud/go-cruise-control/deploy/docker-compose.yml @@ -1,6 +1,6 @@ services: zookeeper: - image: ghcr.io/adobe/zookeeper-operator/zookeeper:3.8.4-0.2.15-adobe-20250923@sha256:170e5b7a8e0156e414f18b9e2c4fa2331f80a3f587bab55ce28577e4d3d642b5 + image: ghcr.io/adobe/zookeeper-operator/zookeeper:3.8.4-0.2.15-adobe-20260818@sha256:ee0fe4635019797be3f55b3efd806dac307b2a95a4041735128c4fbef7494bbb restart: unless-stopped hostname: zookeeper container_name: zookeeper @@ -68,7 +68,7 @@ services: start_period: 30s kafka-0: - image: ghcr.io/adobe/koperator/kafka:2.13-3.9.1@sha256:279358e7bc789aba1e3457421e251278ec993a57d0ad2b9691274f1f3ddae134 + image: ghcr.io/adobe/koperator/kafka:2.13-3.9.2-jdk21.0.11@sha256:80c1f6074e4d305c353cd3f3e31fddceff140da8d4d95baec3ec007702c30b1e restart: unless-stopped user: root hostname: kafka-0 @@ -121,7 +121,7 @@ services: start_period: 30s kafka-1: - image: ghcr.io/adobe/koperator/kafka:2.13-3.9.1@sha256:279358e7bc789aba1e3457421e251278ec993a57d0ad2b9691274f1f3ddae134 + image: ghcr.io/adobe/koperator/kafka:2.13-3.9.2-jdk21.0.11@sha256:80c1f6074e4d305c353cd3f3e31fddceff140da8d4d95baec3ec007702c30b1e restart: unless-stopped user: root hostname: kafka-1 @@ -174,7 +174,7 @@ services: start_period: 30s kafka-2: - image: ghcr.io/adobe/koperator/kafka:2.13-3.9.1@sha256:279358e7bc789aba1e3457421e251278ec993a57d0ad2b9691274f1f3ddae134 + image: ghcr.io/adobe/koperator/kafka:2.13-3.9.2-jdk21.0.11@sha256:80c1f6074e4d305c353cd3f3e31fddceff140da8d4d95baec3ec007702c30b1e restart: unless-stopped user: root hostname: kafka-2 @@ -262,7 +262,7 @@ services: start_period: 30s producer: - image: ghcr.io/adobe/koperator/kafka:2.13-3.9.1@sha256:279358e7bc789aba1e3457421e251278ec993a57d0ad2b9691274f1f3ddae134 + image: ghcr.io/adobe/koperator/kafka:2.13-3.9.2-jdk21.0.11@sha256:80c1f6074e4d305c353cd3f3e31fddceff140da8d4d95baec3ec007702c30b1e restart: unless-stopped hostname: producer container_name: producer @@ -299,7 +299,7 @@ services: start_period: 30s consumer: - image: ghcr.io/adobe/koperator/kafka:2.13-3.9.1@sha256:279358e7bc789aba1e3457421e251278ec993a57d0ad2b9691274f1f3ddae134 + image: ghcr.io/adobe/koperator/kafka:2.13-3.9.2-jdk21.0.11@sha256:80c1f6074e4d305c353cd3f3e31fddceff140da8d4d95baec3ec007702c30b1e restart: unless-stopped hostname: consumer container_name: consumer diff --git a/third_party/github.com/banzaicloud/operator-tools/pkg/reconciler/component.go b/third_party/github.com/banzaicloud/operator-tools/pkg/reconciler/component.go index ae8b38bb8..a4e578e77 100644 --- a/third_party/github.com/banzaicloud/operator-tools/pkg/reconciler/component.go +++ b/third_party/github.com/banzaicloud/operator-tools/pkg/reconciler/component.go @@ -145,7 +145,7 @@ func (r *Dispatcher) Handle(object runtime.Object) (ctrl.Result, error) { if currentDeletionStatus != isBeingDeleted { // Requeue object so that we can start reconciling in inverse order - combinedResult.Combine(&reconcile.Result{Requeue: true, RequeueAfter: 0}, errors.New("object being deleted requeing object")) + combinedResult.CombineErr(errors.New("object being deleted requeing object")) break } diff --git a/third_party/github.com/banzaicloud/operator-tools/pkg/reconciler/resource.go b/third_party/github.com/banzaicloud/operator-tools/pkg/reconciler/resource.go index 4956941aa..c99391b6b 100644 --- a/third_party/github.com/banzaicloud/operator-tools/pkg/reconciler/resource.go +++ b/third_party/github.com/banzaicloud/operator-tools/pkg/reconciler/resource.go @@ -522,7 +522,6 @@ func (r *GenericResourceReconciler) ReconcileResource(desired runtime.Object, de return nil, errors.WrapIfWithDetails(err, "failed to delete current resource", resourceDetails...) } return &reconcile.Result{ - Requeue: true, RequeueAfter: time.Second * time.Duration(utils.PointerToInt32(r.Options.RecreateRequeueDelay)), }, nil } else {