adobe
diff --git a/‎Makefile‎
Lines changed: 1 addition & 1 deletion b/‎Makefile‎
Lines changed: 1 addition & 1 deletion
diff --git a/‎api/go.mod‎
Lines changed: 11 additions & 0 deletions b/‎api/go.mod‎
Lines changed: 11 additions & 0 deletions
diff --git a/‎api/util/util.go‎
Lines changed: 18 additions & 0 deletions b/‎api/util/util.go‎
Lines changed: 18 additions & 0 deletions
diff --git a/‎api/util/util_test.go‎
Lines changed: 10 additions & 0 deletions b/‎api/util/util_test.go‎
Lines changed: 10 additions & 0 deletions
diff --git a/‎api/v1beta1/kafkacluster_types.go‎
Lines changed: 94 additions & 4 deletions b/‎api/v1beta1/kafkacluster_types.go‎
Lines changed: 94 additions & 4 deletions
@@ -102,7 +102,7 @@ test: generate fmt vet bin/setup-envtest
 test-e2e:
 	 IMG_E2E=${IMG_E2E} go test github.com/banzaicloud/koperator/tests/e2e \
 		-v \
-		-timeout 20m \
+		-timeout 45m \
 		-tags e2e \
 		--ginkgo.show-node-events \
 		--ginkgo.trace \
 
@@ -15,6 +15,7 @@ require (
 )
 
 require (
+	// github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
 	github.com/go-logr/logr v1.3.0 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/google/go-cmp v0.5.9 // indirect
@@ -23,16 +24,26 @@ require (
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
+	// github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	golang.org/x/net v0.18.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
 	gopkg.in/inf.v0 v0.9.1 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
+	// gopkg.in/yaml.v3 v3.0.1 // indirect
 	k8s.io/klog/v2 v2.110.1 // indirect
 	k8s.io/utils v0.0.0-20230726121419-3b25d923346b // indirect
 	sigs.k8s.io/json v0.0.0-20221116044647-bc3834ca7abd // indirect
 	sigs.k8s.io/structured-merge-diff/v4 v4.4.1 // indirect
 )
 
+require github.com/stretchr/testify v1.8.4
+
+require (
+	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
+	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
+	gopkg.in/yaml.v3 v3.0.1 // indirect
+)
+
 // remove once https://github.com/cert-manager/cert-manager/issues/5953 is fixed
 replace github.com/Venafi/vcert/v4 => github.com/jetstack/vcert/v4 v4.9.6-0.20230127103832-3aa3dfd6613d
@@ -39,3 +39,21 @@ func MergeLabels(l ...map[string]string) map[string]string {
 func LabelsForKafka(name string) map[string]string {
 	return map[string]string{"app": "kafka", "kafka_cr": name}
 }
+
+func LabelsForBroker(name string) map[string]string {
+	return map[string]string{"isBrokerNode": "true", "app": "kafka", "kafka_cr": name}
+}
+
+func LabelsForController(name string) map[string]string {
+	return map[string]string{"isControllerNode": "true", "app": "kafka", "kafka_cr": name}
+}
+
+// StringSliceContains returns true if list contains s
+func StringSliceContains(list []string, s string) bool {
+	for _, v := range list {
+		if v == s {
+			return true
+		}
+	}
+	return false
+}
@@ -35,3 +35,13 @@ func TestMergeLabels(t *testing.T) {
 		t.Error("Expected:", expected, "Got:", merged)
 	}
 }
+
+func TestStringSliceContains(t *testing.T) {
+	slice := []string{"1", "2", "3"}
+	if !StringSliceContains(slice, "1") {
+		t.Error("Expected slice contains 1, got false")
+	}
+	if StringSliceContains(slice, "4") {
+		t.Error("Expected slice not contains 4, got true")
+	}
+}
@@ -42,6 +42,25 @@ const (
 	KafkaCRLabelKey  = "kafka_cr"
 	BrokerIdLabelKey = "brokerId"
 
+	// ProcessRolesKey is used to identify which process roles the Kafka pod has
+	ProcessRolesKey = "processRoles"
+
+	// IsBrokerNodeKey is used to identify if the kafka pod is either a broker or a broker_controller
+	IsBrokerNodeKey = "isBrokerNode"
+
+	// IsControllerNodeKey is used to identify if the kafka pod is a controller or broker_controller
+	IsControllerNodeKey = "isControllerNode"
+
+	// DefaultCruiseControlImage is the default CC image used when users don't specify it in CruiseControlConfig.Image
+	DefaultCruiseControlImage = "ghcr.io/banzaicloud/cruise-control:2.5.123"
+
+	// DefaultKafkaImage is the default Kafka image used when users don't specify it in KafkaClusterSpec.ClusterImage
+	DefaultKafkaImage = "ghcr.io/banzaicloud/kafka:2.13-3.4.1"
+
+	// ControllerNodeProcessRole represents the node is a controller node
+	ControllerNodeProcessRole = "controller"
+	// BrokerNodeProcessRole represents the node is a broker node
+	BrokerNodeProcessRole = "broker"
 	// These are default values for API keys
 
 	/* General Config */
@@ -134,15 +153,25 @@ const (
 
 // KafkaClusterSpec defines the desired state of KafkaCluster
 type KafkaClusterSpec struct {
+	// kRaft is used to decide where the Kafka cluster is under KRaft mode or ZooKeeper mode.
+	// This is default to be true; if set to false, the Kafka cluster is in ZooKeeper mode.
+	// +kubebuilder:default=false
+	// +optional
+	KRaftMode              bool            `json:"kRaft"`
 	HeadlessServiceEnabled bool            `json:"headlessServiceEnabled"`
 	ListenersConfig        ListenersConfig `json:"listenersConfig"`
 	// Custom ports to expose in the container. Example use case: a custom kafka distribution, that includes an integrated metrics api endpoint
 	AdditionalPorts []corev1.ContainerPort `json:"additionalPorts,omitempty"`
 	// ZKAddresses specifies the ZooKeeper connection string
 	// in the form hostname:port where host and port are the host and port of a ZooKeeper server.
-	ZKAddresses []string `json:"zkAddresses"`
+	// Under ZooKeeper mode, this is a must-have configuration.
+	// And if set under KRaft mode, Koperator ignores this configuration.
+	// +optional
+	ZKAddresses []string `json:"zkAddresses,omitempty"`
 	// ZKPath specifies the ZooKeeper chroot path as part
 	// of its ZooKeeper connection string which puts its data under some path in the global ZooKeeper namespace.
+	// If set under KRaft mode, Koperator ignores this configuration.
+	// +optional
 	ZKPath                      string                  `json:"zkPath,omitempty"`
 	RackAwareness               *RackAwareness          `json:"rackAwareness,omitempty"`
 	ClusterImage                string                  `json:"clusterImage,omitempty"`
@@ -197,6 +226,8 @@ type KafkaClusterStatus struct {
 	RollingUpgrade           RollingUpgradeStatus     `json:"rollingUpgradeStatus,omitempty"`
 	AlertCount               int                      `json:"alertCount"`
 	ListenerStatuses         ListenerStatuses         `json:"listenerStatuses,omitempty"`
+	// ClusterID is a base64-encoded random UUID generated by Koperator to run the Kafka cluster in KRaft mode
+	ClusterID string `json:"clusterID,omitempty"`
 }
 
 // RollingUpgradeStatus defines status of rolling upgrade
@@ -244,11 +275,12 @@ type DisruptionBudgetWithStrategy struct {
 	DisruptionBudget `json:",inline"`
 	// The strategy to be used, either minAvailable or maxUnavailable
 	// +kubebuilder:validation:Enum=minAvailable;maxUnavailable
-	Stategy string `json:"strategy,omitempty"`
+	Strategy string `json:"strategy,omitempty"`
 }
 
 // Broker defines the broker basic configuration
 type Broker struct {
+	// id maps to "node.id" configuration in KRaft mode, and it maps to "broker.id" configuration in ZooKeeper mode.
 	// +kubebuilder:validation:Minimum=0
 	// +kubebuilder:validation:Maximum=65535
 	// +kubebuilder:validation:ExclusiveMaximum=true
@@ -260,6 +292,13 @@ type Broker struct {
 
 // BrokerConfig defines the broker configuration
 type BrokerConfig struct {
+	// processRoles defines the role(s) for this particular Kafka node: "broker", "controller", or both.
+	// This must be set in KRaft mode. If set in ZooKeeper mode, Koperator ignores this configuration.
+	// +kubebuilder:validation:MaxItems=2
+	// +kubebuilder:validation:Items:Type=string
+	// +kubebuilder:validation:Items:Enum=controller;broker
+	// +optional
+	Roles                []string                      `json:"processRoles,omitempty"`
 	Image                string                        `json:"image,omitempty"`
 	MetricsReporterImage string                        `json:"metricsReporterImage,omitempty"`
 	Config               string                        `json:"config,omitempty"`
@@ -991,6 +1030,19 @@ func (bConfig *BrokerConfig) GetTerminationGracePeriod() int64 {
 	return *bConfig.TerminationGracePeriod
 }
 
+// GetStorageMountPaths returns a string with comma-separated storage mount paths that the broker uses
+func (bConfig *BrokerConfig) GetStorageMountPaths() string {
+	var mountPaths string
+	for i, sc := range bConfig.StorageConfigs {
+		if i != len(bConfig.StorageConfigs)-1 {
+			mountPaths += sc.MountPath + ","
+		} else {
+			mountPaths += sc.MountPath
+		}
+	}
+	return mountPaths
+}
+
 // GetNodeSelector returns the node selector for cruise control
 func (cConfig *CruiseControlConfig) GetNodeSelector() map[string]string {
 	return cConfig.NodeSelector
@@ -1051,11 +1103,24 @@ func (bConfig *BrokerConfig) GetBrokerAnnotations() map[string]string {
 }
 
 // GetBrokerLabels returns the labels that are applied to broker pods
-func (bConfig *BrokerConfig) GetBrokerLabels(kafkaClusterName string, brokerId int32) map[string]string {
+func (bConfig *BrokerConfig) GetBrokerLabels(kafkaClusterName string, brokerId int32, kRaftMode bool) map[string]string {
+	var kraftLabels map[string]string
+	if kRaftMode {
+		kraftLabels = map[string]string{
+			ProcessRolesKey:     strings.Join(bConfig.Roles, "_"),
+			IsControllerNodeKey: fmt.Sprintf("%t", bConfig.IsControllerNode()),
+			IsBrokerNodeKey:     fmt.Sprintf("%t", bConfig.IsBrokerNode()),
+		}
+	} else { // in ZK mode -> new labels for backward compatibility for the headless service when going from ZK to KRaft
+		kraftLabels = map[string]string{
+			IsControllerNodeKey: fmt.Sprintf("%t", false),
+			IsBrokerNodeKey:     fmt.Sprintf("%t", true),
+		}
+	}
 	return util.MergeLabels(
 		bConfig.BrokerLabels,
 		util.LabelsForKafka(kafkaClusterName),
-		map[string]string{BrokerIdLabelKey: fmt.Sprintf("%d", brokerId)},
+		kraftLabels, map[string]string{BrokerIdLabelKey: fmt.Sprintf("%d", brokerId)},
 	)
 }
 
@@ -1124,6 +1189,31 @@ func (cConfig *CruiseControlConfig) GetResources() *corev1.ResourceRequirements
 	}
 }
 
+// IsBrokerNode returns true when the broker is a broker node
+func (bConfig *BrokerConfig) IsBrokerNode() bool {
+	return util.StringSliceContains(bConfig.Roles, BrokerNodeProcessRole)
+}
+
+// IsControllerNode returns true when the broker is a controller node
+func (bConfig *BrokerConfig) IsControllerNode() bool {
+	return util.StringSliceContains(bConfig.Roles, ControllerNodeProcessRole)
+}
+
+// IsBrokerOnlyNode returns true when the broker is a broker-only node
+func (bConfig *BrokerConfig) IsBrokerOnlyNode() bool {
+	return bConfig.IsBrokerNode() && !bConfig.IsControllerNode()
+}
+
+// IsControllerOnlyNode returns true when the broker is a controller-only node
+func (bConfig *BrokerConfig) IsControllerOnlyNode() bool {
+	return bConfig.IsControllerNode() && !bConfig.IsBrokerNode()
+}
+
+// IsCombinedNode returns true when the broker is a broker + controller node
+func (bConfig *BrokerConfig) IsCombinedNode() bool {
+	return bConfig.IsBrokerNode() && bConfig.IsControllerNode()
+}
+
 // GetResources returns the broker specific Kubernetes resource
 func (bConfig *BrokerConfig) GetResources() *corev1.ResourceRequirements {
 	if bConfig.Resources != nil {
Original file line number	Diff line number	Diff line change
`@@ -35,3 +35,13 @@ func TestMergeLabels(t *testing.T) {`
`35`	`35`	`t.Error("Expected:", expected, "Got:", merged)`
`36`	`36`	`}`
`37`	`37`	`}`
	`38`	`+`
	`39`	`+func TestStringSliceContains(t *testing.T) {`
	`40`	`+ slice := []string{"1", "2", "3"}`
	`41`	`+ if !StringSliceContains(slice, "1") {`
	`42`	`+ t.Error("Expected slice contains 1, got false")`
	`43`	`+ }`
	`44`	`+ if StringSliceContains(slice, "4") {`
	`45`	`+ t.Error("Expected slice not contains 4, got true")`
	`46`	`+ }`
	`47`	`+}`