KRaft Support

KafkaCluster API changes

Option 1

// ControllerMode holds information about whether the Kafka cluster is in ZooKeeper, or in KRaft mode
type ControllerMode string

KRaftControllerMode ControllerMode = "kraft"
// DefaultControllerMode is the default controller mode that the Kafka cluster is in.
DefaultControllerMode ControllerMode = "zookeeper"

type KafkaClusterSpec struct {
	// ControllerMode specifies the Kafka cluster in either ZooKeeper or KRaft mode.
	// +kubebuilder:validation:Enum=kraft;zookeeper
	// +optional
	ControllerMode              ControllerMode  `json:"controllerMode,omitempty"`
        // other existing fields are intentionally ignored here
	Brokers                     []Broker                `json:"brokers"`
	Controllers                 []Controller            `json:"controllers,omitempty"`
	CombinedNodes               []CombinedNode          `json:"combinedNodes,omitempty"`
}

// Controller defines basic configurations for controllers (in KRaft)
type Controller struct {
	Id               int32             `json:"id"`
	ReadOnlyConfig   string            `json:"readOnlyConfig,omitempty"`
	ControllerConfig *ControllerConfig `json:"controllerConfig"`
}

type ControllerConfig struct {
     // Use the existing BrokerConfig as a blueprint to add/remove corresponding fields from the BrokerConfig
     // reference of BrokerConfig: https://github.com/banzaicloud/koperator/blob/master/api/v1beta1/kafkacluster_types.go#L19
}

// Note: need to find a way to merge the BrokerConfig and ControllerConfig nicely
type CombinedNode struct {
    Id                       int32                     `json:"id"`
    ReadOnlyConfig           string                    `json:"readOnlyConfig,omitempty"`
    BrokerConfig             *BrokerConfig             `json:"brokerConfig,omitempty"`
    ControllerConfig         *ControllerConfig         `json:"controllerConfig,omitempty"`
}

Option 2

// ControllerMode holds information about whether the Kafka cluster is in ZooKeeper, or in KRaft mode
type ControllerMode string

KRaftControllerMode ControllerMode = "kraft"
// DefaultControllerMode is the default controller mode that the Kafka cluster is in.
DefaultControllerMode ControllerMode = "zookeeper"

type KafkaClusterSpec struct {
	// ControllerMode specifies the Kafka cluster in either ZooKeeper or KRaft mode.
	// +kubebuilder:validation:Enum=kraft;zookeeper
	// +optional
	ControllerMode              ControllerMode  `json:"controllerMode,omitempty"`
        // other existing fields are intentionally ignored here
	Brokers                     []Broker                `json:"brokers"`
	Controllers                 []Controller            `json:"controllers,omitempty"`
	CombinedNodes               []CombinedNode          `json:"combinedNodes,omitempty"`
}

// CommonConfig holds the common configurations that are applicable to both the "brokers" and "controllers" (in KRaft term)
// In ZooKeeper-mode, this is just a subset of the old BrokerConfig
type CommonConfig struct {
}


// Note: we don't need to change the Broker struct at all
//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
//	Id                int32         `json:"id"`
//	BrokerConfigGroup string        `json:"brokerConfigGroup,omitempty"`
//	ReadOnlyConfig    string        `json:"readOnlyConfig,omitempty"`
//	BrokerConfig      *BrokerConfig `json:"brokerConfig,omitempty"`
//}


// BrokerConfig defines the broker configurations
type BrokerConfig struct {
	CommonConfig         *CommonConfig         `json:"commonConfig,omitempty"`
	BrokerSpecificConfig *BrokerSpecificConfig `json:"brokerSpecificConfig,omitempty"`
}
type BrokerSpecificConfig struct {
	// applicable fields extracted from original BrokerConfig
}

// Controller defines basic configurations for controllers (in KRaft)
type Controller struct {
	Id               int32             `json:"id"`
	ReadOnlyConfig   string            `json:"readOnlyConfig,omitempty"`
	ControllerConfig *ControllerConfig `json:"controllerConfig"`
}

type ControllerConfig struct { 
    CommonConfig             *CommonConfig             `json:"commonConfig,omitempty"`
	ControllerSpecificConfig *ControllerSpecificConfig `json:"controllerSpecificConfig,omitempty"`
}

// Note: need to find a way to merge the BrokerConfig and ControllerConfig nicely for the CombinedNode
type CombinedNode struct {
    Id                       int32                     `json:"id"`
    ReadOnlyConfig           string                    `json:"readOnlyConfig,omitempty"`
    BrokerConfig             *BrokerConfig             `json:"brokerConfig,omitempty"`
    ControllerConfig         *ControllerConfig         `json:"controllerConfig,omitempty"`
}

Pros and Cons of the two options

	Option 1	Option 2
Pros	No changes required for the BrokerConfig, and therefore no (or minimal) refactoring will be required	Having CommonConfig that is shared between the “Controllers” and “Brokers” seems to make more logical sense
Cons	We will introduce quite some duplicated code: the ControllerConfig is going to have a lot of same fields as the existing BrokerConfig	A lot of changes and refactoring will be required

Idea of managing controller nodes as StatefulSet

Currently Koperator manages all the Kafka brokers as vanilla K8s Pods (along with ConfigMaps to store configurations and PersistentVolumeClaims to store the actual data of the Kafka cluster). The design was to sacrifice simplicity to provide more flexibility to allow users to:

• Modify configurations for unique Brokers
• Remove specific Brokers from the K8s cluster (and therefore the Kafka cluster)
• Use multiple Persistent Volumes for each Broker if needed

The cons of StatefulSet:

• K8s always removes the most recently created Broker pod
• Pods cannot use multiple Persistent Volumes