Add env var to disable creation of PodDisruptionBudget

CHANGELOG.md

@@ -6,8 +6,9 @@
 * The `ContinueReconciliationOnManualRollingUpdateFailure` feature gate moves to beta stage and is enabled by default.
   If needed, `ContinueReconciliationOnManualRollingUpdateFailure` can be disabled in the feature gates configuration in the Cluster Operator.
 * Add support for managing connector offsets via KafkaConnector and KafkaMirrorMaker2 custom resources.
-* Add support for templating `host` and `advertisedHost` fields in listener configuration
-* Allow configuration of environment variables based on Config Map or Secret for every container in the container template sections
+* Add support for templating `host` and `advertisedHost` fields in listener configuration.
+* Allow configuration of environment variables based on Config Map or Secret for every container in the container template sections.
+* Add support for disabling the generation of PodDisruptionBudget resources by the Cluster Operator.
 
 ### Changes, deprecations and removals
 

cluster-operator/src/main/java/io/strimzi/operator/cluster/ClusterOperatorConfig.java

@@ -249,6 +249,11 @@ public class ClusterOperatorConfig {
      */
     /* test */ static final ConfigParameter<String> POD_SECURITY_PROVIDER_RESTRICTED_CLASS = new ConfigParameter<>("POD_SECURITY_PROVIDER_RESTRICTED_CLASS", STRING, "io.strimzi.plugin.security.profiles.impl.RestrictedPodSecurityProvider", CONFIG_VALUES);
 
+    /**
+     * Set true to generate Pod Disruption Budgets
+     */
+    public static final ConfigParameter<Boolean> POD_DISRUPTION_BUDGET_GENERATION = new ConfigParameter<>("STRIMZI_POD_DISRUPTION_BUDGET_GENERATION", BOOLEAN, "true", CONFIG_VALUES);
+
     /**
      * The configured Kafka versions
      */
@@ -598,6 +603,13 @@ public LeaderElectionManagerConfig getLeaderElectionConfig() {
         }
     }
 
+    /**
+     * @return  Indicates whether Pod Disruption Budgets should be generated
+     */
+    public boolean isPodDisruptionBudgetGeneration() {
+        return get(POD_DISRUPTION_BUDGET_GENERATION);
+    }
+
     @Override
     public String toString() {
         return "ClusterOperatorConfig{" +
@@ -620,6 +632,7 @@ public String toString() {
                 "\n\toperatorName='" + getOperatorName() + '\'' +
                 "\n\tpodSecurityProviderClass='" + getPodSecurityProviderClass() + '\'' +
                 "\n\tleaderElectionConfig='" + getLeaderElectionConfig() + '\'' +
+                "\n\tpodDisruptionBudgetGeneration=" + isPodDisruptionBudgetGeneration() +
                 "}";
     }
 }

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/AbstractConnectOperator.java

@@ -101,6 +101,7 @@ public abstract class AbstractConnectOperator<C extends KubernetesClient, T exte
 
     private final boolean isNetworkPolicyGeneration;
     private final boolean continueOnManualRUFailure;
+    private final boolean isPodDisruptionBudgetGeneration;
 
     protected final Function<Vertx, KafkaConnectApi> connectClientProvider;
     protected final ImagePullPolicy imagePullPolicy;
@@ -164,6 +165,7 @@ public AbstractConnectOperator(Vertx vertx, PlatformFeaturesAvailability pfa, St
         this.sharedEnvironmentProvider = supplier.sharedEnvironmentProvider;
         this.port = port;
         this.continueOnManualRUFailure = config.featureGates().continueOnManualRUFailureEnabled();
+        this.isPodDisruptionBudgetGeneration = config.isPodDisruptionBudgetGeneration();
     }
 
     @Override
@@ -299,7 +301,7 @@ protected Future<Void> reconcilePodSet(Reconciliation reconciliation,
                                          KafkaConnectCluster connect,
                                          Map<String, String> podAnnotations,
                                          Map<String, String> podSetAnnotations,
-                                         String customContainerImage)  {
+                                         String customContainerImage) {
         return podSetOperations.reconcile(reconciliation, reconciliation.namespace(), connect.getComponentName(), connect.generatePodSet(connect.getReplicas(), podSetAnnotations, podAnnotations, pfa.isOpenshift(), imagePullPolicy, imagePullSecrets, customContainerImage))
                 .compose(reconciliationResult -> {
                     KafkaConnectRoller roller = new KafkaConnectRoller(reconciliation, connect, operationTimeoutMs, podOperations);
@@ -308,6 +310,24 @@ protected Future<Void> reconcilePodSet(Reconciliation reconciliation,
                 .compose(i -> podSetOperations.readiness(reconciliation, reconciliation.namespace(), connect.getComponentName(), 1_000, operationTimeoutMs));
     }
 
+    /**
+     * Reconciles the PodDisruptionBudget for the Connect cluster.
+     *
+     * @param reconciliation       The reconciliation
+     * @param namespace            Namespace of the Connect cluster
+     * @param connect              KafkaConnectCluster object
+     *
+     * @return                     Future for tracking the asynchronous result of reconciling the PodDisruptionBudget
+     */
+    protected Future<Void> connectPodDisruptionBudget(Reconciliation reconciliation, String namespace, KafkaConnectCluster connect) {
+        if (isPodDisruptionBudgetGeneration) {
+            return podDisruptionBudgetOperator.reconcile(reconciliation, namespace, connect.getComponentName(), connect.generatePodDisruptionBudget())
+                    .mapEmpty();
+        } else {
+            return Future.succeededFuture();
+        }
+    }
+
     /**
      * Dynamically updates loggers in the Kafka Connect cluster.
      *

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/KafkaBridgeAssemblyOperator.java

@@ -48,6 +48,7 @@ public class KafkaBridgeAssemblyOperator extends AbstractAssemblyOperator<Kubern
 
     private final DeploymentOperator deploymentOperations;
     private final SharedEnvironmentProvider sharedEnvironmentProvider;
+    private final boolean isPodDisruptionBudgetGeneration;
 
     /**
      * @param vertx The Vertx instance
@@ -64,6 +65,7 @@ public KafkaBridgeAssemblyOperator(Vertx vertx, PlatformFeaturesAvailability pfa
         super(vertx, pfa, KafkaBridge.RESOURCE_KIND, certManager, passwordGenerator, supplier.kafkaBridgeOperator, supplier, config);
         this.deploymentOperations = supplier.deploymentOperations;
         this.sharedEnvironmentProvider = supplier.sharedEnvironmentProvider;
+        this.isPodDisruptionBudgetGeneration = config.isPodDisruptionBudgetGeneration();
     }
 
     @Override
@@ -97,7 +99,7 @@ protected Future<KafkaBridgeStatus> createOrUpdate(Reconciliation reconciliation
             .compose(scale -> serviceOperations.reconcile(reconciliation, namespace, KafkaBridgeResources.serviceName(bridge.getCluster()), bridge.generateService()))
             .compose(i -> MetricsAndLoggingUtils.metricsAndLogging(reconciliation, configMapOperations, bridge.logging(), null))
             .compose(metricsAndLogging -> configMapOperations.reconcile(reconciliation, namespace, KafkaBridgeResources.metricsAndLogConfigMapName(reconciliation.name()), bridge.generateMetricsAndLogConfigMap(metricsAndLogging)))
-            .compose(i -> podDisruptionBudgetOperator.reconcile(reconciliation, namespace, bridge.getComponentName(), bridge.generatePodDisruptionBudget()))
+            .compose(i -> isPodDisruptionBudgetGeneration ? podDisruptionBudgetOperator.reconcile(reconciliation, namespace, bridge.getComponentName(), bridge.generatePodDisruptionBudget()) : Future.succeededFuture())
             .compose(i -> VertxUtil.authTlsHash(secretOperations, namespace, auth, trustedCertificates))
             .compose(hash -> deploymentOperations.reconcile(reconciliation, namespace, bridge.getComponentName(), bridge.generateDeployment(Collections.singletonMap(Annotations.ANNO_STRIMZI_AUTH_HASH, Integer.toString(hash)), pfa.isOpenshift(), imagePullPolicy, imagePullSecrets)))
             .compose(i -> deploymentOperations.scaleUp(reconciliation, namespace, bridge.getComponentName(), bridge.getReplicas(), operationTimeoutMs))

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/KafkaConnectAssemblyOperator.java

@@ -203,7 +203,7 @@ protected Future<KafkaConnectStatus> createOrUpdate(Reconciliation reconciliatio
                     return configMapOperations.reconcile(reconciliation, namespace, logAndMetricsConfigMap.getMetadata().getName(), logAndMetricsConfigMap);
                 })
                 .compose(i -> ReconcilerUtils.reconcileJmxSecret(reconciliation, secretOperations, connect))
-                .compose(i -> podDisruptionBudgetOperator.reconcile(reconciliation, namespace, connect.getComponentName(), connect.generatePodDisruptionBudget()))
+                .compose(i -> connectPodDisruptionBudget(reconciliation, namespace, connect))
                 .compose(i -> generateAuthHash(namespace, kafkaConnect.getSpec()))
                 .compose(hash -> {
                     podAnnotations.put(Annotations.ANNO_STRIMZI_AUTH_HASH, Integer.toString(hash));

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/KafkaMirrorMaker2AssemblyOperator.java

@@ -141,7 +141,7 @@ protected Future<KafkaMirrorMaker2Status> createOrUpdate(Reconciliation reconcil
                     return configMapOperations.reconcile(reconciliation, namespace, logAndMetricsConfigMap.getMetadata().getName(), logAndMetricsConfigMap);
                 })
                 .compose(i -> ReconcilerUtils.reconcileJmxSecret(reconciliation, secretOperations, mirrorMaker2Cluster))
-                .compose(i -> podDisruptionBudgetOperator.reconcile(reconciliation, namespace, mirrorMaker2Cluster.getComponentName(), mirrorMaker2Cluster.generatePodDisruptionBudget()))
+                .compose(i -> connectPodDisruptionBudget(reconciliation, namespace, mirrorMaker2Cluster))
                 .compose(i -> generateAuthHash(namespace, kafkaMirrorMaker2.getSpec()))
                 .compose(hash -> {
                     podAnnotations.put(Annotations.ANNO_STRIMZI_AUTH_HASH, Integer.toString(hash));

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/KafkaReconciler.java

@@ -114,6 +114,7 @@ public class KafkaReconciler {
     // Various settings
     private final long operationTimeoutMs;
     private final boolean isNetworkPolicyGeneration;
+    private final boolean isPodDisruptionBudgetGeneration;
     private final boolean isKafkaNodePoolsEnabled;
     private final List<String> maintenanceWindows;
     private final String operatorNamespace;
@@ -213,6 +214,7 @@ public KafkaReconciler(
         this.imagePullPolicy = config.getImagePullPolicy();
         this.imagePullSecrets = config.getImagePullSecrets();
         this.previousNodeIds = kafkaCr.getStatus() != null ? kafkaCr.getStatus().getRegisteredNodeIds() : null;
+        this.isPodDisruptionBudgetGeneration = config.isPodDisruptionBudgetGeneration();
 
         this.stsOperator = supplier.stsOperations;
         this.strimziPodSetOperator = supplier.strimziPodSetOperator;
@@ -765,9 +767,13 @@ protected Future<Void> jmxSecret() {
      * @return  Completes when the PDB was successfully created or updated
      */
     protected Future<Void> podDisruptionBudget() {
-        return podDisruptionBudgetOperator
+        if (isPodDisruptionBudgetGeneration) {
+            return podDisruptionBudgetOperator
                     .reconcile(reconciliation, reconciliation.namespace(), KafkaResources.kafkaComponentName(reconciliation.name()), kafka.generatePodDisruptionBudget())
-                    .map((Void) null);
+                    .mapEmpty();
+        } else {
+            return Future.succeededFuture();
+        }
     }
 
     /**

cluster-operator/src/main/java/io/strimzi/operator/cluster/operator/assembly/ZooKeeperReconciler.java

@@ -83,6 +83,7 @@ public class ZooKeeperReconciler {
     private final String operatorNamespace;
     private final Labels operatorNamespaceLabels;
     private final boolean isNetworkPolicyGeneration;
+    private final boolean isPodDisruptionBudgetGeneration;
     private final PlatformFeaturesAvailability pfa;
     private final int adminSessionTimeoutMs;
     private final ImagePullPolicy imagePullPolicy;
@@ -161,6 +162,7 @@ public ZooKeeperReconciler(
         this.imagePullPolicy = config.getImagePullPolicy();
         this.imagePullSecrets = config.getImagePullSecrets();
         this.isKRaftMigrationRollback = isKRaftMigrationRollback;
+        this.isPodDisruptionBudgetGeneration = config.isPodDisruptionBudgetGeneration();
 
         this.stsOperator = supplier.stsOperations;
         this.strimziPodSetOperator = supplier.strimziPodSetOperator;
@@ -492,9 +494,13 @@ protected Future<Void> loggingAndMetricsConfigMap() {
      * @return  Completes when the PDB was successfully created or updated
      */
     protected Future<Void> podDisruptionBudget() {
-        return podDisruptionBudgetOperator
-                .reconcile(reconciliation, reconciliation.namespace(), KafkaResources.zookeeperComponentName(reconciliation.name()), zk.generatePodDisruptionBudget())
-                .map((Void) null);
+        if (isPodDisruptionBudgetGeneration) {
+            return podDisruptionBudgetOperator
+                    .reconcile(reconciliation, reconciliation.namespace(), KafkaResources.zookeeperComponentName(reconciliation.name()), zk.generatePodDisruptionBudget())
+                    .mapEmpty();
+        } else {
+            return Future.succeededFuture();
+        }
     }
 
     /**

cluster-operator/src/test/java/io/strimzi/operator/cluster/ClusterOperatorConfigTest.java

@@ -47,6 +47,7 @@ public class ClusterOperatorConfigTest {
         ENV_VARS.put(ClusterOperatorConfig.FEATURE_GATES.key(), "-ContinueReconciliationOnManualRollingUpdateFailure");
         ENV_VARS.put(ClusterOperatorConfig.DNS_CACHE_TTL.key(), "10");
         ENV_VARS.put(ClusterOperatorConfig.POD_SECURITY_PROVIDER_CLASS.key(), "my.package.CustomPodSecurityProvider");
+        ENV_VARS.put(ClusterOperatorConfig.POD_DISRUPTION_BUDGET_GENERATION.key(), "false");
     }
 
     @Test
@@ -57,6 +58,7 @@ public void testDefaultConfig() {
         envVars.remove(ClusterOperatorConfig.CONNECT_BUILD_TIMEOUT_MS.key());
         envVars.remove(ClusterOperatorConfig.FEATURE_GATES.key());
         envVars.remove(ClusterOperatorConfig.POD_SECURITY_PROVIDER_CLASS.key());
+        envVars.remove(ClusterOperatorConfig.POD_DISRUPTION_BUDGET_GENERATION.key());
 
         ClusterOperatorConfig config = ClusterOperatorConfig.buildFromMap(envVars, KafkaVersionTestUtils.getKafkaVersionLookup());
 
@@ -71,6 +73,7 @@ public void testDefaultConfig() {
         assertThat(config.isPodSetReconciliationOnly(), is(false));
         assertThat(config.getPodSecurityProviderClass(), is(ClusterOperatorConfig.POD_SECURITY_PROVIDER_CLASS.defaultValue()));
         assertThat(config.getLeaderElectionConfig(), is(nullValue()));
+        assertThat(config.isPodDisruptionBudgetGeneration(), is(true));
     }
 
     @Test
@@ -103,6 +106,7 @@ public void testEnvVars() {
         assertThat(config.featureGates().continueOnManualRUFailureEnabled(), is(false));
         assertThat(config.getDnsCacheTtlSec(), is(10));
         assertThat(config.getPodSecurityProviderClass(), is("my.package.CustomPodSecurityProvider"));
+        assertThat(config.isPodDisruptionBudgetGeneration(), is(false));
     }
 
     @Test
@@ -120,6 +124,7 @@ public void testEnvVarsDefault() {
         assertThat(config.featureGates().continueOnManualRUFailureEnabled(), is(true));
         assertThat(config.getDnsCacheTtlSec(), is(Integer.parseInt(ClusterOperatorConfig.DNS_CACHE_TTL.defaultValue())));
         assertThat(config.getPodSecurityProviderClass(), is(ClusterOperatorConfig.POD_SECURITY_PROVIDER_CLASS.defaultValue()));
+        assertThat(config.isPodDisruptionBudgetGeneration(), is(true));
     }
 
     private Map<String, String> envWithImages() {

documentation/assemblies/configuring/assembly-config.adoc

@@ -177,6 +177,9 @@ include::../../modules/configuring/ref-kubernetes-labels.adoc[leveloffset=+1]
 //scheduling separate Kafka pods 
 include::assembly-scheduling.adoc[leveloffset=+1]
 
+//disabling pod disruption budgets
+include::../../modules/configuring/proc-disable-pod-disruption-budget-generation.adoc[leveloffset=+1]
+
 //configuring log levels
 include::assembly-logging-configuration.adoc[leveloffset=+1]
 

documentation/modules/configuring/proc-disable-pod-disruption-budget-generation.adoc

@@ -0,0 +1,11 @@
+// Module included in the following assemblies:
+//
+// assembly-config.adoc
+
+[id='disable-pod-disruption-budget-generation_{context}']
+= Disabling pod disruption budget generation
+
+Strimzi generates pod disruption budget resources for Kafka, Kafka Connect worker, MirrorMaker2 worker, and Kafka Bridge worker nodes.
+
+If you want to use custom pod disruption budget resources, you can set the `STRIMZI_POD_DISRUPTION_BUDGET_GENERATION` environment variable to `false` in the Cluster Operator configuration.
+For more information, see xref:ref-operator-cluster-{context}[].

documentation/modules/operators/con-configuring-cluster-operator.adoc

@@ -80,6 +80,12 @@ env:
     value: label1=value1,label2=value2
 ----
 
+`STRIMZI_POD_DISRUPTION_BUDGET_GENERATION`::  Optional, default `true`.
+Pod disruption budget for resources.
+A pod disruption budget with the `maxUnavailable` value set to zero prevents Kubernetes from evicting pods automatically.
++
+Set this environment variable to `false` to disable pod disruption budget generation. You might do this, for example, if you want to manage the pod disruption budgets yourself, or if you have a development environment where availability is not important.
+
 `STRIMZI_LABELS_EXCLUSION_PATTERN`:: Optional, default regex pattern is `(^app.kubernetes.io/(?!part-of).*|^kustomize.toolkit.fluxcd.io.*)`.
 The regex exclusion pattern used to filter labels propagation from the main custom resource to its subresources.
 The labels exclusion filter is not applied to labels in template sections such as `spec.kafka.template.pod.metadata.labels`.

packaging/helm-charts/helm3/strimzi-kafka-operator/templates/060-Deployment-strimzi-cluster-operator.yaml

@@ -140,6 +140,10 @@ spec:
             - name: STRIMZI_CONNECT_BUILD_TIMEOUT_MS
               value: {{ .Values.connectBuildTimeoutMs | quote }}
             {{- end }}
+            {{- if ne .Values.generatePodDisruptionBudget true}}
+            - name: STRIMZI_POD_DISRUPTION_BUDGET_GENERATION
+              value: {{ .Values.generatePodDisruptionBudget | quote }}
+            {{- end }}
             {{- if .Values.extraEnvs }}
 {{ toYaml .Values.extraEnvs | indent 12 }}
             {{- end }}

packaging/helm-charts/helm3/strimzi-kafka-operator/values.yaml

@@ -180,4 +180,6 @@ labelsExclusionPattern: ""
 # Controls whether Strimzi generates network policy resources (By default true)
 generateNetworkPolicy: true
 # Override the value for Connect build timeout
-connectBuildTimeoutMs: 300000
+connectBuildTimeoutMs: 300000
+# Controls whether Strimzi generates pod disruption budget resources (By default true)
+generatePodDisruptionBudget: true