fix(kafka): address branch-review — restore CA cert secrets, guard KRaft finalize

Follow-up to the KRaft migration addressing the branch review.

- kafka-rd cozyrds: restore the two public CA certs in secrets.include
  (clients-ca-cert + cluster-ca-cert). The rebase had collapsed them to
  the bare clients-ca, which is Strimzi's clients-CA *private key* secret
  — a private-key exposure to the tenant and a TLS regression (clients
  need cluster-ca-cert to trust the broker listener). Covered by the
  existing hack/check-kafka-rd-secrets.bats.

- migration-hook: guard the KRaft finalize step. The wait loop only exited
  via break on success and fell through to unconditionally apply
  strimzi.io/kraft=enabled on timeout — unsafe and irreversible if the
  cluster is still in ZooKeeper/KRaftMigration. Add a migration_ready flag
  and exit 1 when the safe state (KRaftPostMigration+) is not reached,
  leaving the deployed-version ConfigMap unstamped so Helm retries. New
  matchRegex assertions in migration-hook_test.yaml cover the guard.

- migration-hook: raise activeDeadlineSeconds 1800 -> 3600 so a
  legitimately-progressing migration is not SIGKILLed before the in-script
  waits and error paths can run.

- Extract the duplicated controller-quorum logic into a shared
  kafka.controllerReplicas helper (used by kafkanodepools, workloadmonitor
  and migration-hook) so the three stay in lockstep.

- Reflow the README migration section to one line per paragraph per the
  repo prose rule.

Signed-off-by: Alexey Artamonov <aleksei.artamonov@aenix.io>

packages/apps/kafka/README.md

@@ -79,41 +79,24 @@ topics:
 
 ## ZooKeeper to KRaft Migration
 
-The chart itself is now pure KRaft — it ships a Kafka CR with
-`strimzi.io/kraft: enabled` and separate broker + controller `KafkaNodePool`
-resources, with no `spec.zookeeper` block.
+The chart itself is now pure KRaft — it ships a Kafka CR with `strimzi.io/kraft: enabled` and separate broker + controller `KafkaNodePool` resources, with no `spec.zookeeper` block.
 
-Existing ZooKeeper-based instances are migrated automatically on the next chart
-upgrade by a Helm `pre-upgrade` Job, gated by a `<release>-kafka-deployed-version`
-ConfigMap shipped with the chart.
+Existing ZooKeeper-based instances are migrated automatically on the next chart upgrade by a Helm `pre-upgrade` Job, gated by a `<release>-kafka-deployed-version` ConfigMap shipped with the chart.
 
 ### How it works
 
 1. The Job renders only when the version ConfigMap is missing or stamped below `"1"`.
-2. On upgrade, it inspects the existing Kafka CR's `status.kafkaMetadataState`:
-   - If the CR is absent (fresh install) or already in `KRaft`, it exits immediately.
-   - Otherwise (typically `ZooKeeper` state), it creates the broker + controller
-     `KafkaNodePool` resources matching the chart's values and annotates the
-     Kafka CR with `strimzi.io/node-pools=enabled` and `strimzi.io/kraft=migration`.
-3. The Job polls `status.kafkaMetadataState` and waits for the migration to
-   reach `KRaftPostMigration | PreKRaft | KRaft`.
-4. It then flips the annotation to `strimzi.io/kraft=enabled` and waits until
-   the state reaches `KRaft`.
-5. When the Job succeeds, Helm applies the chart's KRaft manifests (which match
-   the post-migration state) and stamps the ConfigMap to `"1"`. Subsequent
-   reconciles see the ConfigMap and skip the Job entirely.
-6. If the Job fails or times out, Helm aborts the upgrade — the ConfigMap stays
-   below the threshold, so the next reconcile re-runs the same Job.
+2. On upgrade, it inspects the existing Kafka CR's `status.kafkaMetadataState`: if the CR is absent (fresh install) or already in `KRaft` it exits immediately; otherwise (typically `ZooKeeper` state) it creates the broker + controller `KafkaNodePool` resources matching the chart's values and annotates the Kafka CR with `strimzi.io/node-pools=enabled` and `strimzi.io/kraft=migration`.
+3. The Job polls `status.kafkaMetadataState` and waits for the migration to reach `KRaftPostMigration | PreKRaft | KRaft`.
+4. Only once that safe state is reached does it flip the annotation to `strimzi.io/kraft=enabled` and wait until the state reaches `KRaft`; if the wait times out in an intermediate state the Job aborts (`exit 1`) without finalising, leaving the ConfigMap unstamped.
+5. When the Job succeeds, Helm applies the chart's KRaft manifests (which match the post-migration state) and stamps the ConfigMap to `"1"`. Subsequent reconciles see the ConfigMap and skip the Job entirely.
+6. If the Job fails or times out, Helm aborts the upgrade — the ConfigMap stays below the threshold, so the next reconcile re-runs the same Job.
 
 ### Observability and escape hatches
 
-- Tail the Job logs to follow migration progress:
-  `kubectl logs -n <namespace> job/<release>-kafka-migration`
+- Tail the Job logs to follow migration progress: `kubectl logs -n <namespace> job/<release>-kafka-migration`
 - Monitor `status.kafkaMetadataState` on the Kafka CR directly.
-- If migration gets stuck before `KRaftPostMigration`, Strimzi's `rollback`
-  annotation stays available as a manual escape hatch:
-  `kubectl annotate kafka <release> strimzi.io/kraft=rollback --overwrite`,
-  then delete the failed Job and retry.
+- If migration gets stuck before `KRaftPostMigration`, Strimzi's `rollback` annotation stays available as a manual escape hatch: `kubectl annotate kafka <release> strimzi.io/kraft=rollback --overwrite`, then delete the failed Job and retry.
 
 ### Important notes
 

packages/apps/kafka/templates/_versions.tpl

@@ -5,3 +5,13 @@
 {{- end }}
 {{- index $versionMap .Values.version }}
 {{- end }}
+
+{{- /*
+  KRaft controller quorum: an odd number of controllers. Single-node clusters
+  (kafka.replicas <= 1) run one controller; everything else runs three. Shared
+  by kafkanodepools.yaml, workloadmonitor.yaml and migration-hook.yaml so the
+  three stay in lockstep.
+*/ -}}
+{{- define "kafka.controllerReplicas" -}}
+{{- if le (int .Values.kafka.replicas) 1 -}}1{{- else -}}3{{- end -}}
+{{- end -}}

packages/apps/kafka/templates/kafkanodepools.yaml

@@ -1,8 +1,4 @@
-{{- /* Controller replicas: odd quorum (1 for single-node, 3 otherwise) */ -}}
-{{- $controllerReplicas := 3 }}
-{{- if le (int .Values.kafka.replicas) 1 }}
-  {{- $controllerReplicas = 1 }}
-{{- end }}
+{{- $controllerReplicas := include "kafka.controllerReplicas" . }}
 ---
 apiVersion: kafka.strimzi.io/v1beta2
 kind: KafkaNodePool

packages/apps/kafka/templates/migration-hook.yaml

@@ -6,10 +6,7 @@
   {{- end }}
 {{- end }}
 {{- if $shouldRun }}
-{{- $controllerReplicas := 3 }}
-{{- if le (int .Values.kafka.replicas) 1 }}
-  {{- $controllerReplicas = 1 }}
-{{- end }}
+{{- $controllerReplicas := include "kafka.controllerReplicas" . }}
 ---
 apiVersion: v1
 kind: ServiceAccount
@@ -60,7 +57,10 @@ metadata:
     helm.sh/hook-delete-policy: hook-succeeded,before-hook-creation
 spec:
   backoffLimit: 0
-  activeDeadlineSeconds: 1800
+  # Worst-case in-script waits total ~1800s of sleeps (240x5s + 120x5s) before
+  # kubectl round-trip latency; keep headroom above that so a genuinely-progressing
+  # migration is not SIGKILLed mid-flight and the guard/error paths can run.
+  activeDeadlineSeconds: 3600
   template:
     metadata:
       labels:
@@ -155,11 +155,13 @@ spec:
               kubectl -n "$ns" annotate kafka "$name" strimzi.io/kraft=migration --overwrite
 
               echo "Waiting for migration to reach KRaftPostMigration..."
+              migration_ready=false
               for i in $(seq 1 240); do
                 state=$(kubectl -n "$ns" get kafka "$name" -o jsonpath='{.status.kafkaMetadataState}' || echo "")
                 case "$state" in
                   KRaftPostMigration|PreKRaft|KRaft)
                     echo "Reached state: $state"
+                    migration_ready=true
                     break
                     ;;
                   *)
@@ -169,6 +171,19 @@ spec:
                 esac
               done
 
+              # Strimzi only permits strimzi.io/kraft=enabled once the cluster has
+              # reached KRaftPostMigration. Applying it earlier — while brokers still
+              # depend on ZooKeeper — is unsafe and irreversible (kraft=rollback is no
+              # longer available past that point). If the wait timed out in an
+              # intermediate state, abort without finalising and without stamping the
+              # deployed-version ConfigMap, so Helm fails the upgrade and the next
+              # reconcile retries from where the migration left off.
+              if [ "$migration_ready" != "true" ]; then
+                echo "ERROR: migration did not reach KRaftPostMigration (last state: ${state:-unknown})."
+                echo "Refusing to apply strimzi.io/kraft=enabled in an unsafe state."
+                exit 1
+              fi
+
               echo "Finalising migration to KRaft..."
               kubectl -n "$ns" annotate kafka "$name" strimzi.io/kraft=enabled --overwrite
 

packages/apps/kafka/templates/workloadmonitor.yaml

@@ -1,7 +1,4 @@
-{{- $controllerReplicas := 3 }}
-{{- if le (int .Values.kafka.replicas) 1 }}
-  {{- $controllerReplicas = 1 }}
-{{- end }}
+{{- $controllerReplicas := include "kafka.controllerReplicas" . }}
 ---
 apiVersion: cozystack.io/v1alpha1
 kind: WorkloadMonitor

packages/apps/kafka/tests/migration-hook_test.yaml

@@ -81,3 +81,26 @@ tests:
           path: rules[0].verbs
           content: patch
         documentIndex: 1
+
+  # The migration Job must never apply strimzi.io/kraft=enabled unless the
+  # cluster actually reached KRaftPostMigration — flipping it earlier is
+  # unsafe and irreversible. Guard the finalize step behind an explicit
+  # readiness flag that aborts (exit 1) on timeout, leaving the deployed-version
+  # ConfigMap unstamped so Helm retries.
+  - it: guards the KRaft finalize step behind a reached-safe-state check
+    release:
+      name: test-kafka
+      namespace: tenant-test
+    asserts:
+      - matchRegex:
+          path: spec.template.spec.containers[0].args[0]
+          pattern: "migration_ready=false"
+        documentIndex: 3
+      - matchRegex:
+          path: spec.template.spec.containers[0].args[0]
+          pattern: 'if \[ "\$migration_ready" != "true" \]; then'
+        documentIndex: 3
+      - matchRegex:
+          path: spec.template.spec.containers[0].args[0]
+          pattern: "Refusing to apply strimzi.io/kraft=enabled in an unsafe state"
+        documentIndex: 3

packages/system/kafka-rd/cozyrds/kafka.yaml

@@ -30,7 +30,8 @@ spec:
     exclude: []
     include:
       - resourceNames:
-          - kafka-{{ .name }}-clients-ca
+          - kafka-{{ .name }}-clients-ca-cert
+          - kafka-{{ .name }}-cluster-ca-cert
   services:
     exclude: []
     include: