<a id="co-kraft"></a>

# Configure and Manage KRaft for Confluent Platform Using Confluent for Kubernetes

In Confluent for Kubernetes (CFK), you can use Kafka Raft (KRaft) for Kafka
consensus and metadata storage. This document describes the configuration for
KRaft in CFK.

KRaft utilizes the [Raft](https://raft.github.io/) consensus algorithm and
is part of Kafka. KRaft stores metadata within a Kafka metadata topic. For
detail information, see [KRaft Overview](https://docs.confluent.io/platform/current/kafka-metadata/kraft.html).

KRaft in CFK works with Confluent Platform 7.4.0 and later.

In CFK, you can configure, deploy, and manage KRaft with a declarative API.
CFK provides a custom resource definition (CRD) for the KRaft controller
(KRaftController), and the CR configures the following:

KRaft controller
: KRaft uses controllers for metadata consensus and storage. An odd number (at
  least 3) of KRaft controller replicas are required.

KRaft controller listener
: Communication to and among the KRaft controller nodes happens over the
  controller listener.
  <br/>
  When KRaft is enabled, the controller listener must be identically
  configured on the controller side and broker side, including the same
  authentication and TLS  properties.
  <br/>
  CFK configures the controller listener on port 9074.

The high-level workflow to deploy Confluent Platform with KRaft is:

1. [Deploy CFK](co-deploy-operator.md#co-deploy-operator).

   Alternatively, you can [deploy CFK with the data recovery feature
   enabled](co-cfk-settings.md#co-deployment-kraft). This option does not support namespace-scoped
   deployment because CFK needs to create the ClusterRole and
   ClusterRolebinding required for the feature.
2. [Create KRaft controllers](#co-configure-kraft-controller).
3. [Add a reference to the KRaft controller in the Kafka CR](#co-configure-kraft-kafka).

<a id="co-configure-kraft-controller"></a>

## Configure KRaft controllers

To create a KRaft controller, create and configure a KRaftController CR. The
following shows the key CR settings:

```yaml
kind: KRaftController
metadata:
  name:                                    --- [1]
  namespace:                               --- [2]
spec:
  replicas:                                --- [3]
  listeners:
    controller:                            --- [4]
      authentication:                      --- [5]
      tls:
        enabled:                           --- [6]
      externalAccess:                      --- [7]
        type:                              --- [8]
  controllerQuorumVoters:                  --- [9]
  configOverrides:                         --- [10]
    server:
      - default.replication.factor=        --- [11]
      - listener                           --- [12]
  dependencies:
    schemaRegistry:                        --- [13]
```

* [1] Required. The name of this KRaft controller.
* [2] The namespace of this KRaft controller.
* [3] The desired number of replicas. Must be an odd number that is 3 or higher.

  A change to this setting will roll the cluster.
* [4] Required. Communication to and among the KRaft controller nodes happens
  over this controller listener.
* [5] See [Authentication for Kafka and KRaft](co-authenticate-kafka.md#co-authenticate-kafka) for
  configuring authentication.
* [6] Set to `true` to enable TLS. See [Network encryption](co-network-encryption.md#co-network-encryption) for
  configuring TLS certificates.
* [7] defines the external access configuration for the Kafka cluster.
* [8] Required if `externalAccess` ([7]) is specified. Set to the Kubernetes
  service for external access. Valid options are `loadBalancer`, `nodePort`,
  `route`, `staticForPortBasedRouting`, and `staticForHostBasedRouting`.

  For details on external access configuration, see
  [Network encryption](co-networking-overview.md#co-networking-overview).
* [9] Required for multi-region deployment. Follow the further configuration
  steps in [Set up MRC with KRaft](co-multi-region.md#co-mrc-kraft).
* [10] Required. Use the `configOverrides` to set the matching properties as
  set in the Kafka CR. The following properties are required: [11],
  [12]
* [11] Required. The default replication factor is the number of Kafka replicas.
  This parameter is needed for the KRaft controller to interact with the
  Kafka brokers for some features, such as Self-Balancing and metrics reporter. You must
  explicitly set it for KRaft to match the number of Kafka replicas
  (`spec.replicas` in the Kafka CR).

  Use `configOverrides` to set the property because the property is not
  directly supported in the KRaft CR. For example:
  ```yaml
  spec:
    configOverrides:
      server:
        - default.replication.factor=3
  ```
* [12] Required for when authentication is enabled for Kafka replication
  listeners. Set the same properties set in the Kafka CR, under
  `spec.listeners.replication.authentication`.

  The following are sample configurations for mTLS authentication among
  replication listeners:
  ```yaml
  spec:
    configOverrides:
      server:
        - listener.name.replication.ssl.client.auth=required
        - listener.name.replication.ssl.key.password=${file:/vault/secrets/kafka-tls/jksPassword.txt:jksPassword}
        - listener.name.replication.ssl.keystore.location=/vault/secrets/kafka-tls/keystore.jks
        - listener.name.replication.ssl.keystore.password=${file:/vault/secrets/kafka-tls/jksPassword.txt:jksPassword}
        - listener.name.replication.ssl.principal.mapping.rules=RULE:.*CN[\s]?=[\s]?([a-zA-Z0-9._]*)?.*/$1/
        - listener.name.replication.ssl.truststore.location=/vault/secrets/kafka-tls/truststore.jks
        - listener.name.replication.ssl.truststore.password=${file:/vault/secrets/kafka-tls/jksPassword.txt:jksPassword}
        - listener.security.protocol.map=CONTROLLER:SSL,REPLICATION:SSL
  ```
* [13] Required when the Kafka CR has a dependency on Schema Registry in the
  `spec.dependencies.schemaRegistry` section. Set the same Schema Registry dependency
  settings you set in the Kafka CR here.

An example KRaftController CR:

```yaml
kind: KRaftController
metadata:
  name: kcontroller
  namespace: operator
spec:
  replicas: 3
  listeners:
    controller:
      authentication:
        type: plain
        jaasConfig:
          secretRef: kraft-secret
      tls:
        enabled: true
  dependencies:
    schemaRegistry:
      authentication:
        basic:
          secretRef: kafka-sr-credential
        type: basic
      tls:
        enabled: true
  configOverrides:
    server:
      - default.replication.factor=3
```

<a id="co-configure-kraft-kafka"></a>

## Configure Kafka for KRaft

For KRaft-enabled Kafka, add a kRaftController cluster reference in the
`dependencies` section of the Kafka CR.

```yaml
kind: Kafka
spec:
  dependencies:
    kRaftController:      --- [1]
      clusterRef:         --- [2]
        name:             --- [3]
        namespace:        --- [4]
      controllerListener:
        authentication:   --- [5]
        tls:              --- [6]
```

* [1] Kafka will fail to start if both `zookeeper` and `kRaftController` are
  specified in the `dependencies` object.
* [2] A reference to a kRaftController CR. If omitted, Kafka will automatically
  discover a KRaft controller running in the same namespace.
* [3] The name of the kRaftController CR.
* [4] The namespace of the kRaftController CR.
* [5] Required if the controller listener on the KRaft controller cluster
  (kRaftController CR) has authentication configured. Must be identically
  configured as in the kRaftController CR.
  See [Configure KRaft controllers](#co-configure-kraft-controller).
* [6] Required if the controller listener on the KRaft controller cluster
  (kRaftController CR) has TLS enabled. Must be identically
  configured as in the kRaftController CR. See
  [Configure KRaft controllers](#co-configure-kraft-controller).

The following is an example CR of a KRaft-enabled Kafka:

```yaml
apiVersion: platform.confluent.io/v1beta1
kind: Kafka
metadata:
  name: kafka
  namespace: operator
  annotations:
    platform.confluent.io/broker-id-offset: 10
spec:
  dependencies:
    kRaftController:
      clusterRef:
        name: kcontroller
        namespace: operator

      controllerListener:
        authentication:
          type: plain
          jaasConfig:
            secretRef: secret
        tls:
          enabled: true
```