Kraft

controllers/tests/kafkacluster_controller_kafka_test.go

@@ -88,6 +88,14 @@ func expectKafkaAllBrokerService(ctx context.Context, kafkaCluster *v1beta1.Kafk
 }
 
 func expectKafkaPDB(ctx context.Context, kafkaCluster *v1beta1.KafkaCluster) {
+	if kafkaCluster.Spec.KRaftMode {
+		expectKafkaPDBKraft(ctx, kafkaCluster)
+	} else {
+		expectKafkaPDBZK(ctx, kafkaCluster)
+	}
+}
+
+func expectKafkaPDBZK(ctx context.Context, kafkaCluster *v1beta1.KafkaCluster) {
 	// get current CR
 	err := k8sClient.Get(ctx, types.NamespacedName{Name: kafkaCluster.Name, Namespace: kafkaCluster.Namespace}, kafkaCluster)
 	Expect(err).NotTo(HaveOccurred())
@@ -106,7 +114,7 @@ func expectKafkaPDB(ctx context.Context, kafkaCluster *v1beta1.KafkaCluster) {
 	// wait until reconcile finishes
 	waitForClusterRunningState(ctx, kafkaCluster, kafkaCluster.Namespace)
 
-	// get created PDB
+	// get created broker PDB
 	pdb := policyv1.PodDisruptionBudget{}
 	Eventually(ctx, func() error {
 		return k8sClient.Get(ctx, types.NamespacedName{
@@ -115,13 +123,78 @@ func expectKafkaPDB(ctx context.Context, kafkaCluster *v1beta1.KafkaCluster) {
 		}, &pdb)
 	}).Should(Succeed())
 
-	// make assertions
+	// make assertions for broker pdb
 	Expect(pdb.Labels).To(HaveKeyWithValue(v1beta1.AppLabelKey, "kafka"))
 	Expect(pdb.Labels).To(HaveKeyWithValue(v1beta1.KafkaCRLabelKey, kafkaCluster.Name))
+	Expect(pdb.Labels).To(Not(HaveKey(v1beta1.IsBrokerNodeKey)))
 	Expect(pdb.Spec.MinAvailable).To(Equal(util.IntstrPointer(3)))
 	Expect(pdb.Spec.Selector).NotTo(BeNil())
 	Expect(pdb.Spec.Selector.MatchLabels).To(HaveKeyWithValue(v1beta1.AppLabelKey, "kafka"))
 	Expect(pdb.Spec.Selector.MatchLabels).To(HaveKeyWithValue(v1beta1.KafkaCRLabelKey, kafkaCluster.Name))
+	Expect(pdb.Spec.Selector.MatchLabels).To(Not(HaveKey(v1beta1.IsBrokerNodeKey)))
+
+	Eventually(ctx, func() error {
+		return k8sClient.Get(ctx, types.NamespacedName{
+			Namespace: kafkaCluster.Namespace,
+			Name:      fmt.Sprintf("%s-controller-pdb", kafkaCluster.Name),
+		}, &pdb)
+	}).WithTimeout(1000).Should(MatchError(fmt.Sprintf("poddisruptionbudgets.policy \"%s-controller-pdb\" not found", kafkaCluster.Name)))
+}
+
+func expectKafkaPDBKraft(ctx context.Context, kafkaCluster *v1beta1.KafkaCluster) {
+	// get current CR
+	err := k8sClient.Get(ctx, types.NamespacedName{Name: kafkaCluster.Name, Namespace: kafkaCluster.Namespace}, kafkaCluster)
+	Expect(err).NotTo(HaveOccurred())
+
+	// set PDB and reset status
+	kafkaCluster.Spec.DisruptionBudget = v1beta1.DisruptionBudget{
+		Create: true,
+		Budget: "20%",
+	}
+	kafkaCluster.Status = v1beta1.KafkaClusterStatus{}
+
+	// update CR
+	err = k8sClient.Update(ctx, kafkaCluster)
+	Expect(err).NotTo(HaveOccurred())
+
+	// wait until reconcile finishes
+	waitForClusterRunningState(ctx, kafkaCluster, kafkaCluster.Namespace)
+
+	// get created broker PDB
+	pdb := policyv1.PodDisruptionBudget{}
+	Eventually(ctx, func() error {
+		return k8sClient.Get(ctx, types.NamespacedName{
+			Namespace: kafkaCluster.Namespace,
+			Name:      fmt.Sprintf("%s-pdb", kafkaCluster.Name),
+		}, &pdb)
+	}).Should(Succeed())
+
+	// make assertions for broker pdb
+	Expect(pdb.Labels).To(HaveKeyWithValue(v1beta1.AppLabelKey, "kafka"))
+	Expect(pdb.Labels).To(HaveKeyWithValue(v1beta1.KafkaCRLabelKey, kafkaCluster.Name))
+	Expect(pdb.Labels).To(HaveKeyWithValue(v1beta1.IsBrokerNodeKey, "true"))
+	Expect(pdb.Spec.MinAvailable).To(Equal(util.IntstrPointer(2)))
+	Expect(pdb.Spec.Selector).NotTo(BeNil())
+	Expect(pdb.Spec.Selector.MatchLabels).To(HaveKeyWithValue(v1beta1.AppLabelKey, "kafka"))
+	Expect(pdb.Spec.Selector.MatchLabels).To(HaveKeyWithValue(v1beta1.KafkaCRLabelKey, kafkaCluster.Name))
+	Expect(pdb.Spec.Selector.MatchLabels).To(HaveKeyWithValue(v1beta1.IsBrokerNodeKey, "true"))
+
+	Eventually(ctx, func() error {
+		return k8sClient.Get(ctx, types.NamespacedName{
+			Namespace: kafkaCluster.Namespace,
+			Name:      fmt.Sprintf("%s-controller-pdb", kafkaCluster.Name),
+		}, &pdb)
+	}).Should(Succeed())
+
+	// make assertions for controller pdb
+	Expect(pdb.Labels).To(HaveKeyWithValue(v1beta1.AppLabelKey, "kafka"))
+	Expect(pdb.Labels).To(HaveKeyWithValue(v1beta1.KafkaCRLabelKey, kafkaCluster.Name))
+	Expect(pdb.Labels).To(HaveKeyWithValue(v1beta1.IsControllerNodeKey, "true"))
+	Expect(pdb.Spec.MinAvailable).To(Equal(util.IntstrPointer(1)))
+	Expect(pdb.Spec.Selector).NotTo(BeNil())
+	Expect(pdb.Spec.Selector.MatchLabels).To(HaveKeyWithValue(v1beta1.AppLabelKey, "kafka"))
+	Expect(pdb.Spec.Selector.MatchLabels).To(HaveKeyWithValue(v1beta1.KafkaCRLabelKey, kafkaCluster.Name))
+	Expect(pdb.Spec.Selector.MatchLabels).To(HaveKeyWithValue(v1beta1.IsControllerNodeKey, "true"))
 }
 
 func expectKafkaPVC(ctx context.Context, kafkaCluster *v1beta1.KafkaCluster) {

hack/boilerplate/header.generated.txt

@@ -1,4 +1,4 @@
-Copyright 2024 Cisco Systems, Inc. and/or its affiliates
+Copyright 2025 Cisco Systems, Inc. and/or its affiliates
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.

hack/boilerplate/header.go.generated.txt

@@ -1,5 +1,5 @@
 /*
-Copyright 2024 Cisco Systems, Inc. and/or its affiliates
+Copyright 2025 Cisco Systems, Inc. and/or its affiliates
 
 Licensed under the Apache License, Version 2.0 (the "License");
 you may not use this file except in compliance with the License.

pkg/resources/kafka/kafka.go

@@ -238,11 +238,23 @@ func (r *Reconciler) Reconcile(log logr.Logger) error {
 		}
 	}
 
-	// Handle PDB
+	// Handle PDB for brokers
 	if r.KafkaCluster.Spec.DisruptionBudget.Create {
-		o, err := r.podDisruptionBudget(log)
+		o, err := r.podDisruptionBudgetBrokers(log)
 		if err != nil {
-			return errors.WrapIfWithDetails(err, "failed to compute podDisruptionBudget")
+			return errors.WrapIfWithDetails(err, "failed to compute podDisruptionBudget for brokers")
+		}
+		err = k8sutil.Reconcile(log, r.Client, o, r.KafkaCluster)
+		if err != nil {
+			return errors.WrapIfWithDetails(err, "failed to reconcile resource", "resource", o.GetObjectKind().GroupVersionKind())
+		}
+	}
+
+	// Handle PDB for controllers (KRaft only)
+	if r.KafkaCluster.Spec.KRaftMode && r.KafkaCluster.Spec.DisruptionBudget.Create {
+		o, err := r.podDisruptionBudgetControllers(log)
+		if err != nil {
+			return errors.WrapIfWithDetails(err, "failed to compute podDisruptionBudget for controllers")
 		}
 		err = k8sutil.Reconcile(log, r.Client, o, r.KafkaCluster)
 		if err != nil {

pkg/resources/kafka/poddisruptionbudget.go

@@ -15,6 +15,7 @@
 package kafka
 
 import (
+	"errors"
 	"fmt"
 	"math"
 	"strconv"
@@ -32,33 +33,92 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 )
 
-func (r *Reconciler) podDisruptionBudget(log logr.Logger) (runtime.Object, error) {
+func (r *Reconciler) podDisruptionBudgetBrokers(log logr.Logger) (runtime.Object, error) {
+	var podSelectorLabels map[string]string
 	minAvailable, err := r.computeMinAvailable(log)
 
 	if err != nil {
 		return nil, err
 	}
 
+	if r.KafkaCluster.Spec.KRaftMode {
+		podSelectorLabels = apiutil.LabelsForBroker(r.KafkaCluster.Name)
+	} else {
+		podSelectorLabels = apiutil.LabelsForKafka(r.KafkaCluster.Name)
+	}
+
+	return r.podDisruptionBudget(fmt.Sprintf("%s-pdb", r.KafkaCluster.Name),
+		podSelectorLabels,
+		minAvailable)
+}
+
+func (r *Reconciler) podDisruptionBudgetControllers(log logr.Logger) (runtime.Object, error) {
+	if !r.KafkaCluster.Spec.KRaftMode {
+		return nil, errors.New("PDB for controllers is only applicable when in KRaft mode")
+	}
+
+	minAvailable, err := r.computeControllerMinAvailable()
+
+	if err != nil {
+		log.Error(err, "error occurred during computing minAvailable for controllers PDB")
+		return nil, err
+	}
+
+	return r.podDisruptionBudget(fmt.Sprintf("%s-controller-pdb", r.KafkaCluster.Name),
+		apiutil.LabelsForController(r.KafkaCluster.Name),
+		minAvailable)
+}
+
+func (r *Reconciler) podDisruptionBudget(name string, podSelectorLabels map[string]string, minAvailable intstr.IntOrString) (runtime.Object, error) {
 	return &policyv1.PodDisruptionBudget{
 		TypeMeta: metav1.TypeMeta{
 			Kind:       "PodDisruptionBudget",
 			APIVersion: "policy/v1",
 		},
 		ObjectMeta: templates.ObjectMetaWithAnnotations(
-			fmt.Sprintf("%s-pdb", r.KafkaCluster.Name),
-			apiutil.MergeLabels(apiutil.LabelsForKafka(r.KafkaCluster.Name), r.KafkaCluster.Labels),
+			name,
+			apiutil.MergeLabels(podSelectorLabels, r.KafkaCluster.Labels),
 			r.KafkaCluster.Spec.ListenersConfig.GetServiceAnnotations(),
 			r.KafkaCluster,
 		),
 		Spec: policyv1.PodDisruptionBudgetSpec{
 			MinAvailable: &minAvailable,
 			Selector: &metav1.LabelSelector{
-				MatchLabels: apiutil.LabelsForKafka(r.KafkaCluster.Name),
+				MatchLabels: podSelectorLabels,
 			},
 		},
 	}, nil
 }
 
+func (r *Reconciler) getControllerCount(controllerRoleOnly bool) (int, error) {
+	controllerCount := 0
+	for _, broker := range r.KafkaCluster.Spec.Brokers {
+		brokerConfig, err := broker.GetBrokerConfig(r.KafkaCluster.Spec)
+		if err != nil {
+			return -1, err
+		}
+		if controllerRoleOnly {
+			if brokerConfig.IsControllerOnlyNode() {
+				controllerCount++
+			}
+		} else if brokerConfig.IsControllerNode() {
+			controllerCount++
+		}
+	}
+	return controllerCount, nil
+}
+
+// Calculate minAvailable as max between brokerCount - 1 (so we only allow 1 controller to be disrupted)
+// and 1 (case when there is only 1 controller)
+func (r *Reconciler) computeControllerMinAvailable() (intstr.IntOrString, error) {
+	controllerCount, err := r.getControllerCount(false)
+	if err != nil {
+		return intstr.FromInt(-1), err
+	}
+	minAvailable := int(math.Max(float64(controllerCount-1), float64(1)))
+	return intstr.FromInt(minAvailable), nil
+}
+
 // Calculate maxUnavailable as max between brokerCount - 1 (so we only allow 1 broker to be disrupted)
 // and 1 (to cover for 1 broker clusters)
 func (r *Reconciler) computeMinAvailable(log logr.Logger) (intstr.IntOrString, error) {
@@ -73,8 +133,15 @@ func (r *Reconciler) computeMinAvailable(log logr.Logger) (intstr.IntOrString, e
 		Max(1, brokers-brokers*percentage) - for a percentage budget
 
 	*/
-	// number of brokers in the KafkaCluster
-	brokers := len(r.KafkaCluster.Status.BrokersState)
+
+	controllerCount, err := r.getControllerCount(true)
+	if err != nil {
+		log.Error(err, "error occurred during get controller count")
+		return intstr.FromInt(-1), err
+	}
+
+	// number of brokers in the KafkaCluster.  Controllers are reported in the BrokerState so we must deduct it.
+	brokers := len(r.KafkaCluster.Status.BrokersState) - controllerCount
 
 	// configured budget in the KafkaCluster
 	disruptionBudget := r.KafkaCluster.Spec.DisruptionBudget.Budget