@@ -60,7 +60,7 @@ case "$args" in
     [ -n "${STUB_CREATE_FAIL:-}" ] && { echo "AlreadyExists" >&2; exit 1; }
     exit 0
     ;;
-  *"apply -f -"*)
+  *"apply "*"-f -"*)
     cat >> "${KUBECTL_LOG:-/dev/null}"
     exit 0
     ;;
@@ -206,7 +206,7 @@ run_hook() {
 
   [ "$rc" != 0 ]
   grep -qF 'already belongs to cluster' "$tmp/out"
-  if grep -qE 'apply -f -|create -f -|kraft=|node-pools=' "$tmp/kubectl.log"; then
+  if grep -qE 'apply .*-f -|create -f -|kraft=|node-pools=' "$tmp/kubectl.log"; then
     echo "FAIL: a foreign-owned kafka pool did not abort before mutating"
     cat "$tmp/kubectl.log"
     return 1
@@ -223,7 +223,7 @@ run_hook() {
 
   [ "$rc" != 0 ]
   grep -qF 'owner label failed' "$tmp/out"
-  if grep -qE 'apply -f -|create -f -|kraft=|node-pools=' "$tmp/kubectl.log"; then
+  if grep -qE 'apply .*-f -|create -f -|kraft=|node-pools=' "$tmp/kubectl.log"; then
     echo "FAIL: proceeded to mutate the cluster after a failed owner read"
     cat "$tmp/kubectl.log"
     return 1
 
@@ -0,0 +1,20 @@
+{{- /*
+  JAVA_TOOL_OPTIONS for a Strimzi-managed container, from its resource limits.
+  The JDK 17 in the Strimzi images cannot see the container limits on kernels
+  built without CONFIG_MEMCG_V1 (JDK-8347811) and sizes itself from the host:
+  Strimzi's heap probe then hands every broker up to a 5Gi heap. Every java in
+  the container reads this variable, the probe included, so Strimzi's own
+  heap rules work again. jvmOptions -XX would not do: Strimzi moves those into
+  KAFKA_JVM_PERFORMANCE_OPTS, and kafka-run-class.sh drops its G1 tuning when
+  that variable is set. ActiveProcessorCount only when the chart sets a CPU
+  limit: a default a LimitRange adds at admission is not visible here, and
+  Strimzi template env takes no resourceFieldRef, so such a JVM still counts
+  the host CPUs.
+*/ -}}
+{{- define "kafka.javaToolOptions" -}}
+{{- $opts := list (printf "-XX:MaxRAM=%d" (include "cozy-lib.resources.toFloat" .memory | float64 | int64)) -}}
+{{- with .cpu -}}
+{{- $opts = append $opts (printf "-XX:ActiveProcessorCount=%d" (include "cozy-lib.resources.toFloat" . | float64 | ceil | int64)) -}}
+{{- end -}}
+{{- join " " $opts -}}
+{{- end -}}
@@ -16,6 +16,7 @@
 {{- $tlsEnabled := include "kafka.tls.enabled" . | trim -}}
 {{- $showExternal := .Values.external -}}
 {{- $kafkaVersion := include "kafka.versionMap" . -}}
+{{- $entityOperatorMemory := "512Mi" -}}
 apiVersion: kafka.strimzi.io/v1beta2
 kind: Kafka
 metadata:
@@ -75,21 +76,22 @@ spec:
     # The topic-operator and user-operator are JVM processes and OOMKill at
     # that default, leaving the entity-operator in CrashLoopBackOff so
     # KafkaTopics/KafkaUsers never reconcile. Set explicit resources so they
-    # don't inherit the namespace default.
+    # don't inherit the namespace default. JAVA_TOOL_OPTIONS: see
+    # kafka.javaToolOptions.
     topicOperator:
       resources:
         requests:
           cpu: 100m
           memory: 256Mi
         limits:
-          memory: 512Mi
+          memory: {{ $entityOperatorMemory }}
     userOperator:
       resources:
         requests:
           cpu: 100m
           memory: 256Mi
         limits:
-          memory: 512Mi
+          memory: {{ $entityOperatorMemory }}
     template:
       # `pod` accepts metadata + a flat set of pod-spec fields
       # (enableServiceLinks, securityContext, affinity, …) directly.
@@ -101,3 +103,11 @@ spec:
           labels:
             policy.cozystack.io/allow-to-apiserver: "true"
         enableServiceLinks: false
+      topicOperatorContainer:
+        env:
+          - name: JAVA_TOOL_OPTIONS
+            value: {{ include "kafka.javaToolOptions" (dict "memory" $entityOperatorMemory) | quote }}
+      userOperatorContainer:
+        env:
+          - name: JAVA_TOOL_OPTIONS
+            value: {{ include "kafka.javaToolOptions" (dict "memory" $entityOperatorMemory) | quote }}
@@ -26,6 +26,7 @@
 {{- $controllerReplicas := include "kafka.controllerReplicas" . }}
 {{- $brokerPool := include "kafka.brokerPoolName" . }}
 {{- $freshBrokerPool := printf "b-%s" (include "kafka.poolHash" .) }}
+{{- $cruiseControlMemory := "1Gi" }}
 ---
 apiVersion: v1
 kind: ServiceAccount
@@ -121,6 +122,15 @@ spec:
               np=kafkanodepools.kafka.strimzi.io
               kr=kafkarebalances.kafka.strimzi.io
 
+              # Every apply here passes --openapi-patch=false. Without it, applying
+              # an object that already exists makes kubectl download the cluster's
+              # whole OpenAPI v2 document only to learn that a custom resource
+              # takes a JSON merge patch: about 300Mi with the platform's CRDs,
+              # while this container runs under a tenant LimitRange default of
+              # 128Mi. With it kubectl builds the same merge patch from the
+              # last-applied annotation. Not --server-side: fields that
+              # "kubectl create" set on the "kafka" pool stay co-owned by that
+              # manager, so a server-side apply would never remove them.
               # ---- Node-pool manifests (broker pool reused for the create-CAS) ----
               broker_manifest=$(cat <<MANIFEST
               apiVersion: kafka.strimzi.io/v1beta2
@@ -143,6 +153,14 @@ spec:
                 roles:
                   - broker
                 resources: {{- include "cozy-lib.resources.defaultingSanitize" (list .Values.kafka.resourcesPreset .Values.kafka.resources $) | nindent 18 }}
+                # A pool template replaces spec.kafka.template whole, for every
+                # pool here: nothing set on the Kafka CR's template reaches these
+                # pods (KafkaPool.java at 0.45.1-rc1: https://github.com/strimzi/strimzi-kafka-operator/blob/0.45.1-rc1/cluster-operator/src/main/java/io/strimzi/operator/cluster/model/KafkaPool.java#L196-L218).
+                template:
+                  kafkaContainer:
+                    env:
+                      - name: JAVA_TOOL_OPTIONS
+                        value: {{ include "kafka.javaToolOptions" (include "cozy-lib.resources.defaultingSanitize" (list .Values.kafka.resourcesPreset .Values.kafka.resources $) | fromYaml).limits | quote }}
                 storage:
                   type: jbod
                   volumes:
@@ -187,6 +205,11 @@ spec:
                 roles:
                   - controller
                 resources: {{- include "cozy-lib.resources.defaultingSanitize" (list .Values.kafka.resourcesPreset .Values.kafka.resources $) | nindent 18 }}
+                template:
+                  kafkaContainer:
+                    env:
+                      - name: JAVA_TOOL_OPTIONS
+                        value: {{ include "kafka.javaToolOptions" (include "cozy-lib.resources.defaultingSanitize" (list .Values.kafka.resourcesPreset .Values.kafka.resources $) | fromYaml).limits | quote }}
                 storage:
                   type: jbod
                   volumes:
@@ -242,10 +265,10 @@ spec:
                 fi
               fi
               echo "Ensuring broker KafkaNodePool \"$brokerPool\" exists..."
-              echo "$broker_manifest" | kubectl apply -f -
+              echo "$broker_manifest" | kubectl apply --openapi-patch=false -f -
 
               echo "Ensuring controller KafkaNodePool exists..."
-              echo "$controller_manifest" | kubectl apply -f -
+              echo "$controller_manifest" | kubectl apply --openapi-patch=false -f -
 
               # ---- Decide what to do from the Kafka CR ----
               # Fail closed. --ignore-not-found makes kubectl exit 0 with empty
@@ -370,7 +393,7 @@ spec:
               # F1: stand up this cluster's own release-hashed broker pool with
               # node IDs pinned high, clear of the existing broker/controller IDs.
               echo "Creating target broker pool \"$freshBrokerPool\"..."
-              cat <<POOL | kubectl apply -f -
+              cat <<POOL | kubectl apply --openapi-patch=false -f -
               apiVersion: kafka.strimzi.io/v1beta2
               kind: KafkaNodePool
               metadata:
@@ -388,6 +411,11 @@ spec:
                 roles:
                   - broker
                 resources: {{- include "cozy-lib.resources.defaultingSanitize" (list .Values.kafka.resourcesPreset .Values.kafka.resources $) | nindent 18 }}
+                template:
+                  kafkaContainer:
+                    env:
+                      - name: JAVA_TOOL_OPTIONS
+                        value: {{ include "kafka.javaToolOptions" (include "cozy-lib.resources.defaultingSanitize" (list .Values.kafka.resourcesPreset .Values.kafka.resources $) | fromYaml).limits | quote }}
                 storage:
                   type: jbod
                   volumes:
@@ -407,9 +435,13 @@ spec:
               # explicit resources: Cruise Control is a JVM, and a tenant namespace
               # LimitRange defaults containers to 128Mi, which OOMs it before it
               # can produce a proposal (same reason the entity-operator carries
-              # explicit resources in kafka.yaml).
+              # explicit resources in kafka.yaml). JAVA_TOOL_OPTIONS: see
+              # kafka.javaToolOptions.
               echo "Enabling Cruise Control for the drain..."
-              kubectl -n "$ns" patch "$kafka" "$name" --type merge -p '{"spec":{"cruiseControl":{"resources":{"requests":{"cpu":"200m","memory":"768Mi"},"limits":{"memory":"1Gi"}}}}}'
+              {{- $ccEnv := list (dict "name" "JAVA_TOOL_OPTIONS" "value" (include "kafka.javaToolOptions" (dict "memory" $cruiseControlMemory))) }}
+              {{- $ccResources := dict "requests" (dict "cpu" "200m" "memory" "768Mi") "limits" (dict "memory" $cruiseControlMemory) }}
+              {{- $ccPatch := dict "spec" (dict "cruiseControl" (dict "resources" $ccResources "template" (dict "cruiseControlContainer" (dict "env" $ccEnv)))) }}
+              kubectl -n "$ns" patch "$kafka" "$name" --type merge -p '{{ $ccPatch | toJson }}'
 
               # F3: drain all partitions off the "kafka" pool broker IDs.
               kafkaIds=$(kubectl -n "$ns" get "$np" kafka -o jsonpath='{.status.nodeIds}' 2>/dev/null || echo "")
@@ -418,7 +450,7 @@ spec:
                 exit 1
               fi
               echo "Draining brokers $kafkaIds via Cruise Control rebalance..."
-              cat <<REB | kubectl apply -f -
+              cat <<REB | kubectl apply --openapi-patch=false -f -
               apiVersion: kafka.strimzi.io/v1beta2
               kind: KafkaRebalance
               metadata:
 
@@ -0,0 +1,97 @@
+suite: Kafka JVMs are sized from their container limits
+
+templates:
+  - templates/migration-hook.yaml
+  - templates/kafka.yaml
+
+release:
+  name: test-kafka
+  namespace: tenant-test
+
+tests:
+  - it: passes the default preset limits to every node pool JVM
+    template: templates/migration-hook.yaml
+    documentIndex: 3
+    asserts:
+      - matchRegex:
+          path: spec.template.spec.containers[0].args[0]
+          pattern: "release-namespace: \\$\\{ns\\}\\s+spec:\\s+replicas: \\d+\\s+roles:\\s+- broker\\s+resources:\\s+limits:(\\s+\\S+: \\S+)+\\s+requests:(\\s+\\S+: \\S+)+(\\s+#[^\\n]*)*\\s+template:\\s+kafkaContainer:\\s+env:\\s+- name: JAVA_TOOL_OPTIONS\\s+value: \"-XX:MaxRAM=1073741824 -XX:ActiveProcessorCount=1\"\\s+storage:"
+      - matchRegex:
+          path: spec.template.spec.containers[0].args[0]
+          pattern: "roles:\\s+- controller\\s+resources:\\s+limits:(\\s+\\S+: \\S+)+\\s+requests:(\\s+\\S+: \\S+)+(\\s+#[^\\n]*)*\\s+template:\\s+kafkaContainer:\\s+env:\\s+- name: JAVA_TOOL_OPTIONS\\s+value: \"-XX:MaxRAM=1073741824 -XX:ActiveProcessorCount=1\"\\s+storage:"
+      - matchRegex:
+          path: spec.template.spec.containers[0].args[0]
+          pattern: "next-node-ids: \"\\[1000-1099\\]\"\\s+spec:\\s+replicas: \\d+\\s+roles:\\s+- broker\\s+resources:\\s+limits:(\\s+\\S+: \\S+)+\\s+requests:(\\s+\\S+: \\S+)+(\\s+#[^\\n]*)*\\s+template:\\s+kafkaContainer:\\s+env:\\s+- name: JAVA_TOOL_OPTIONS\\s+value: \"-XX:MaxRAM=1073741824 -XX:ActiveProcessorCount=1\"\\s+storage:"
+
+  # Strimzi moves jvmOptions -XX into KAFKA_JVM_PERFORMANCE_OPTS, and
+  # kafka-run-class.sh then drops its G1 settings.
+  - it: leaves the node pool JVM options to Strimzi and Kafka
+    template: templates/migration-hook.yaml
+    documentIndex: 3
+    asserts:
+      - notMatchRegex:
+          path: spec.template.spec.containers[0].args[0]
+          pattern: "jvmOptions"
+
+  - it: rounds a fractional CPU limit up
+    template: templates/migration-hook.yaml
+    documentIndex: 3
+    set:
+      kafka.resources:
+        cpu: 2500m
+        memory: 3Gi
+    asserts:
+      - matchRegex:
+          path: spec.template.spec.containers[0].args[0]
+          pattern: "roles:\\s+- controller\\s+resources:\\s+limits:(\\s+\\S+: \\S+)+\\s+requests:(\\s+\\S+: \\S+)+(\\s+#[^\\n]*)*\\s+template:\\s+kafkaContainer:\\s+env:\\s+- name: JAVA_TOOL_OPTIONS\\s+value: \"-XX:MaxRAM=3221225472 -XX:ActiveProcessorCount=3\"\\s+storage:"
+
+  - it: takes the limits, not the requests, when the allocation ratios shrink the requests
+    template: templates/migration-hook.yaml
+    documentIndex: 3
+    set:
+      _cluster:
+        cpu-allocation-ratio: "4"
+        memory-allocation-ratio: "2"
+      kafka.resources:
+        cpu: "2"
+        memory: 2Gi
+    asserts:
+      - matchRegex:
+          path: spec.template.spec.containers[0].args[0]
+          pattern: "release-namespace: \\$\\{ns\\}\\s+spec:\\s+replicas: \\d+\\s+roles:\\s+- broker\\s+resources:\\s+limits:(\\s+\\S+: \\S+)+\\s+requests:(\\s+\\S+: \\S+)+(\\s+#[^\\n]*)*\\s+template:\\s+kafkaContainer:\\s+env:\\s+- name: JAVA_TOOL_OPTIONS\\s+value: \"-XX:MaxRAM=2147483648 -XX:ActiveProcessorCount=2\"\\s+storage:"
+      - matchRegex:
+          path: spec.template.spec.containers[0].args[0]
+          pattern: "roles:\\s+- controller\\s+resources:\\s+limits:(\\s+\\S+: \\S+)+\\s+requests:(\\s+\\S+: \\S+)+(\\s+#[^\\n]*)*\\s+template:\\s+kafkaContainer:\\s+env:\\s+- name: JAVA_TOOL_OPTIONS\\s+value: \"-XX:MaxRAM=2147483648 -XX:ActiveProcessorCount=2\"\\s+storage:"
+      - matchRegex:
+          path: spec.template.spec.containers[0].args[0]
+          pattern: "next-node-ids: \"\\[1000-1099\\]\"\\s+spec:\\s+replicas: \\d+\\s+roles:\\s+- broker\\s+resources:\\s+limits:(\\s+\\S+: \\S+)+\\s+requests:(\\s+\\S+: \\S+)+(\\s+#[^\\n]*)*\\s+template:\\s+kafkaContainer:\\s+env:\\s+- name: JAVA_TOOL_OPTIONS\\s+value: \"-XX:MaxRAM=2147483648 -XX:ActiveProcessorCount=2\"\\s+storage:"
+
+  # The whole patch, closing brace included: a merge patch kubectl cannot
+  # parse fails the hook before the drain starts.
+  - it: passes the drain-time Cruise Control memory limit to its JVM
+    template: templates/migration-hook.yaml
+    documentIndex: 3
+    asserts:
+      - matchRegex:
+          path: spec.template.spec.containers[0].args[0]
+          pattern: "\\-\\-type\\ merge\\ \\-p\\ '\\{\"spec\":\\{\"cruiseControl\":\\{\"resources\":\\{\"limits\":\\{\"memory\":\"1Gi\"\\},\"requests\":\\{\"cpu\":\"200m\",\"memory\":\"768Mi\"\\}\\},\"template\":\\{\"cruiseControlContainer\":\\{\"env\":\\[\\{\"name\":\"JAVA_TOOL_OPTIONS\",\"value\":\"\\-XX:MaxRAM=1073741824\"\\}\\]\\}\\}\\}\\}\\}'\n"
+
+  - it: passes the entity operator memory limit to its JVMs
+    template: templates/kafka.yaml
+    asserts:
+      - equal:
+          path: spec.entityOperator.topicOperator.resources.limits.memory
+          value: 512Mi
+      - equal:
+          path: spec.entityOperator.template.topicOperatorContainer.env
+          value:
+            - name: JAVA_TOOL_OPTIONS
+              value: -XX:MaxRAM=536870912
+      - equal:
+          path: spec.entityOperator.userOperator.resources.limits.memory
+          value: 512Mi
+      - equal:
+          path: spec.entityOperator.template.userOperatorContainer.env
+          value:
+            - name: JAVA_TOOL_OPTIONS
+              value: -XX:MaxRAM=536870912
@@ -447,3 +447,16 @@ tests:
           path: spec.template.spec.containers[0].args[0]
           pattern: "roles:\\s+- controller[\\s\\S]*?resources:\\s+limits:"
         documentIndex: 3
+
+  - it: applies without the OpenAPI patch lookup, so an upgrade fits the tenant LimitRange default
+    release:
+      name: test-kafka
+      namespace: tenant-test
+    documentIndex: 3
+    asserts:
+      - notMatchRegex:
+          path: spec.template.spec.containers[0].args[0]
+          pattern: "kubectl[^|\\n]* apply (-f|--filename|--server-side)"
+      - matchRegex:
+          path: spec.template.spec.containers[0].args[0]
+          pattern: "echo \"\\$broker_manifest\" \\| kubectl apply --openapi-patch=false -f -\\n[\\s\\S]*echo \"\\$controller_manifest\" \\| kubectl apply --openapi-patch=false -f -\\n[\\s\\S]*<<POOL \\| kubectl apply --openapi-patch=false -f -\\n[\\s\\S]*<<REB \\| kubectl apply --openapi-patch=false -f -\\n"
@@ -3,6 +3,9 @@ export NAMESPACE=cozy-$(NAME)
 
 include ../../../hack/package.mk
 
+test:
+	helm unittest .
+
 update:
 	rm -rf charts
 	helm pull oci://quay.io/strimzi-helm/strimzi-kafka-operator --untar --untardir charts --version 0.45.1-rc1
@@ -0,0 +1,31 @@
+suite: cluster operator JVM is sized from the container limits
+release:
+  name: kafka-operator
+  namespace: cozy-kafka-operator
+templates:
+  - charts/strimzi-kafka-operator/templates/060-Deployment-strimzi-cluster-operator.yaml
+tests:
+  - it: passes the container limits to the JVM, which cannot read them on kernels without cgroup v1 controllers
+    asserts:
+      # $(VAR) expands only from entries earlier in the list.
+      - equal:
+          path: spec.template.spec.containers[0].env[19]
+          value:
+            name: CONTAINER_MEMORY_LIMIT
+            valueFrom:
+              resourceFieldRef:
+                resource: limits.memory
+      - equal:
+          path: spec.template.spec.containers[0].env[20]
+          value:
+            name: CONTAINER_CPU_LIMIT
+            valueFrom:
+              resourceFieldRef:
+                resource: limits.cpu
+      - equal:
+          path: spec.template.spec.containers[0].env[21]
+          value:
+            name: JAVA_OPTS
+            value: -XX:MaxRAM=$(CONTAINER_MEMORY_LIMIT) -XX:ActiveProcessorCount=$(CONTAINER_CPU_LIMIT)
+      - exists:
+          path: spec.template.spec.containers[0].resources.limits.cpu
@@ -5,3 +5,19 @@ strimzi-kafka-operator:
   resources:
     limits:
       memory: 512Mi
+  # The JDK 17 in the operator image decides whether to read the container
+  # limits by looking at /proc/cgroups, which loses its memory line on kernels
+  # built without CONFIG_MEMCG_V1 (6.12+). The JVM then sizes its heap from
+  # host RAM and the container is OOMKilled. Drop this once the image ships a
+  # JDK with JDK-8347811 (17.0.20, 21.0.10).
+  extraEnvs:
+    - name: CONTAINER_MEMORY_LIMIT
+      valueFrom:
+        resourceFieldRef:
+          resource: limits.memory
+    - name: CONTAINER_CPU_LIMIT
+      valueFrom:
+        resourceFieldRef:
+          resource: limits.cpu
+    - name: JAVA_OPTS
+      value: -XX:MaxRAM=$(CONTAINER_MEMORY_LIMIT) -XX:ActiveProcessorCount=$(CONTAINER_CPU_LIMIT)