# KRaft Configuration for Confluent Platform This document covers hardware recommendations, configuration, debugging tools, and monitoring options for running Apache Kafka® in KRaft (pronounced craft) mode. ## Hardware and JVM requirements A production KRaft server can cover a wide variety of use cases. In general, you should run KRaft on a server with similar specifications to the following: - At least 4 GB of RAM - Dedicated CPU core should be considered when the server is shared - An SSD disk at least 64 GB in size is highly recommended - JVM heap size of at least 1 GB is recommended #### NOTE [Ansible Playbooks for Confluent Platform](https://docs.confluent.io/ansible/current/overview.html) set a default maximum JVM heap size (`-Xmx`) of 4 GB for KRaft controllers. To avoid out-of-memory errors, provision controller nodes with more than the minimum 4 GB of RAM so that the OS and other processes have memory available outside of the JVM heap. Currently, it is recommended that you run at least three (3) KRaft controllers in production. For more details, see [Hardware](../installation/system-requirements.md#cp-hardware). ## Configuration options Consider that a KRaft controller is also a Kafka broker processing event records that contain metadata related to the Kafka cluster. However, not all broker properties need to be set on controllers. There are some settings that must be included for a cluster to run in KRaft mode, and are unique per server, but there are other settings that you configure for a controller because the controller process itself uses that property to perform its controller duties, or the property affects cluster metadata that controllers manage. For a full list of configuration properties, see [Kafka Broker and Controller Configuration Reference for Confluent Platform](../installation/configuration/broker-configs.md#cp-config-brokers). Settings for KRaft mode are listed in the following sections with links to the configuration reference for those properties. ### Required settings These entries must be included for each server (controllers and brokers) running in KRaft mode. [process.roles](/platform/current/installation/configuration/broker-configs.html#process-roles) : When you operate Apache Kafka® in KRaft mode, you must set the `process.roles` property. This property specifies whether the server acts as a controller, broker, or both, although currently both is not supported for production workloads. In KRaft mode, specific Kafka servers are selected to be controllers, storing metadata for the cluster in the metadata log, and other servers are selected to be brokers. The servers selected to be controllers will participate in the metadata quorum. Each controller is either an active or a hot standby for the current active controller.
In a production environment, the controller quorum will be deployed on multiple nodes. This is called an ensemble. An ensemble is a set of 2n + 1 controllers where n is any number greater than 0. The odd number of controllers allows the controller quorum to perform majority elections for leadership. At any given time, there can be up to n failed servers in an ensemble and cluster will keep quorum. For example, with three controllers, the cluster can tolerate one controller failure. If at any time, quorum is lost, the cluster will go down. For production, you should have typically have 3 or 5 controllers, but at least 3. For more information, see [Hardware](../installation/system-requirements.md#cp-hardware).
- Type: string - Default: - Importance: required for KRaft mode
`process.roles` can have the following values:
| Value | Result | |---------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | Not set | The server is assumed to be in ZooKeeper mode. This is not supported in Confluent Platform version 8.0 and later. | | `broker` | The server operates only as a broker. | | `controller` | The server operates in isolated mode as a controller only. | | `broker,controller` | The server operates in combined mode, where it is both a broker and a controller.
Combined mode is for local experimentation only and is not supported by Confluent.
For an example of combined mode, see the [confluent-local](../installation/docker/config-reference.md#confluent-local-example) Docker image. | [node.id](/platform/current/installation/configuration/broker-configs.html#node-id) : The unique identifier for this server. Each node ID must be unique across all the brokers and controllers in a particular cluster. No two servers can have the same node ID regardless of their `process.roles` value. This identifier replaces `broker.id`, which was used when operating in ZooKeeper mode.
- Type: int - Default: - Importance: required for KRaft mode [controller.quorum.bootstrap.servers](/platform/current/installation/configuration/broker-configs.html#controller-quorum-bootstrap-servers) : **Recommended.** A comma-separated list of controller endpoints for dynamic controller configurations. All servers, including controllers and brokers, in a Kafka cluster use this property to discover the controller quorum.
Specify each controller endpoint with host and port information in the format `{host}:{port}`. Separate multiple entries with commas, as shown in the following example:
`controller.quorum.bootstrap.servers=controller1.example.com:9093,controller2.example.com:9093,controller3.example.com:9093`
Unlike `controller.quorum.voters`, this property does not need to list every controller. List enough controllers to let nodes locate the quorum. However, include as many controller endpoints as feasible for reliability.
For new KRaft clusters, use dynamic controller configuration. Starting with Confluent Platform 8.2, set `controller.quorum.auto.join.enable=true` on all controllers. Then, format the first controller with `--standalone`, and format each additional controller with `--no-initial-controllers`. Controllers automatically join the quorum without manual configuration. For instructions, see [Provision multiple controllers for a new cluster](#provision-multiple-controllers).
If you have an existing cluster with static controllers, upgrade to dynamic controllers. To upgrade, see [Upgrade from static controllers to dynamic controllers](#upgrade-static-to-dynamic-controllers).
- Type: string - Default: - Importance: required for dynamic controller configurations [controller.quorum.voters](/platform/current/installation/configuration/broker-configs.html#controller-quorum-voters) : A comma-separated list of quorum voters for static controller configurations. All servers, including controllers and brokers, in a Kafka cluster use this property to discover the quorum voters when you use static controllers.
Identify each controller with ID, host, and port information in the format `{id}@{host}:{port}`. Separate multiple entries with commas, as shown in the following example:
`controller.quorum.voters=1@host1:port1,2@host2:port2,3@host3:port3`
The node ID in the `controller.quorum.voters` property must match the corresponding ID on the controller servers. For example, on controller1, set `node.id` to `1`. If a server is a broker only, its node ID must not appear in the `controller.quorum.voters` list.
#### NOTE Use `controller.quorum.voters` for static controller configurations. For new clusters, use dynamic controllers with `controller.quorum.bootstrap.servers` instead. Static controllers require you to update configuration on all nodes and restart them to make changes to the controller set.
See [Upgrade from static controllers to dynamic controllers](#upgrade-static-to-dynamic-controllers) to upgrade from static to dynamic controllers.
- Type: string - Default: - Importance: required for static controller configurations [controller.quorum.auto.join.enable](/platform/current/installation/configuration/broker-configs.html#controller-quorum-auto-join-enable) : When set to `true`, KRaft controllers automatically join the cluster’s voter set without requiring manual configuration updates on existing nodes. This removes the need to manually update the voter configuration on existing nodes when adding new controllers. This configuration applies only to dynamic controller configurations using `kraft.version=1` or later and has no effect on static quorum configurations.
Set this only on nodes that specify `controller` for `process.roles`.
- Type: boolean - Default: `false` - Importance: medium [controller.listener.names](/platform/current/installation/configuration/broker-configs.html#controller-listener-names) : A comma-separated list of `listener_name` entries for listeners used by the controller. On a node with `process.roles=broker`, only the first listener in the list will be used by the broker. For KRaft controllers in isolated or combined mode, the node will listen as a KRaft controller on all listeners that are listed for this property, and each must appear in the `listeners` property. They shouldn’t appear in the `advertised.listeners` property.
- Type: string - Default: null - Importance: required ### Inter-broker listeners Listeners are an important part of your configuration. In addition to `controller.listener.names` described in the previous section, you should configure how KRaft controllers will communicate with brokers. This can be done with the `security.inter.broker.protocol` property or the `inter.broker.listener.name` property, but not both. If `inter.broker.listener.name` is set then it will be used as a key for lookup in the `listener.security.protocol.map` property to yield a security protocol, otherwise `security.inter.broker.protocol` will be used. The default for `security.inter.broker.protocol` is `PLAINTEXT`, which is what will be used for communication with brokers if neither property is explicitly set. Note that controllers do not listen at the `inter.broker.listener.name` value, but this property defines a listener that the brokers create, and controllers must specify in their security protocol and configuration so it can communicate with the brokers. Following are descriptions of these properties: [inter.broker.listener.name](/platform/current/installation/configuration/broker-configs.html#inter-broker-listener-name) : The listener name that is used for inter-broker communication. If this is not set, inter-broker communication is defined by the `security.inter.broker.protocol` property. Set one of these, but not both, or an error will occur. This property must be set on KRaft brokers, but note that you must also set this property for KRaft controllers because controllers sometimes need to talk to Kafka brokers in Confluent Platform.
The inter-broker listener name for a controller node **must not** appear in `controller.listener.names` property, and this applies regardless of whether the node is a controller in isolated or combined mode. The following example configuration file for a KRaft controller uses dynamic quorum, the recommended approach, and shows how to configure this property and the security map for the listener.
```bash process.roles=controller node.id=100 controller.quorum.bootstrap.servers=node1:9093,node2:9093,node3:9093 controller.listener.names=CONTROLLER listeners=CONTROLLER://:9093 inter.broker.listener.name=BROKER listener.security.protocol.map=CONTROLLER:SSL,BROKER:SSL
# Define the controller's listener and how we will use it. listener.name.controller.ssl.keystore.location=/some/keystore/path listener.name.controller.ssl.truststore.location=/some/truststore/path # etc...
# Define how we will use the broker's listener. # No keystore needed since the controller isn't listening here; only need a truststore. listener.name.broker.ssl.truststore.location=/some/truststore/path # etc... ```
#### NOTE This example uses `controller.quorum.bootstrap.servers`, the recommended approach for dynamic controller quorums. If you use static controllers, replace this with `controller.quorum.voters=100@node1:9093,101@node2:9093,102@node3:9093` instead. For more information, see [Upgrade from static controllers to dynamic controllers](#upgrade-static-to-dynamic-controllers). [listener.security.protocol.map](/platform/current/installation/configuration/broker-configs.html#listener-security-protocol-map) : The security protocol to use for inter-broker communication specified by the `inter.broker.listener.name` property. The security protocol to use for the declared listener names. Note that this includes controller-to-broker communication with the listener identified by the `inter.broker.listener.name` property for the controller. [security.inter.broker.protocol](/platform/current/installation/configuration/broker-configs.html#security-inter-broker-protocol) : Security protocol used to communicate between brokers. Set this property or `inter.broker.listener.name`, but not both. ### Other listeners and logs Following are additional properties you should be familiar with. [listeners](/platform/current/installation/configuration/broker-configs.html#listeners) : A comma-separated list of addresses where the socket server listens.
For controllers in isolated mode: Only controller listeners are allowed in this list when `process.roles=controller`, and this listener should be consistent with `controller.quorum.voters` value. If not configured, the host name will be equal to the value of `java.net.InetAddress.getCanonicalHostName()` with the `PLAINTEXT` listener name, and port `9092`.
For controllers in combined mode, you should list the controller listeners as well as the broker listeners. For brokers: see [listeners](/platform/current/installation/configuration/broker-configs.html#listeners).
- Type: string with the format `listener_name://host_name:port` - Default: If not configured, the host name will be equal to the value of `java.net.InetAddress.getCanonicalHostName()`, with `PLAINTEXT` listener name, and port `9092`. Example: `listeners=PLAINTEXT://your.host.name:9092` - Importance: high [metadata.log.dir](/platform/current/installation/configuration/broker-configs.html#metadata-log-dir) : Use to specify where the metadata log for clusters in KRaft mode is placed after storage is formatted as described in [Generate and format IDs](#generate-format-ids). If not set, the metadata log is placed in the first log directory specified in the `log-dirs` property described below.
- Type: string - Default: null - Importance: high [log.dirs](/platform/current/installation/configuration/broker-configs.html#log-dirs) : If `metadata.log.dir` is not specified, the KRaft metadata log is placed in the first log directory specified by this property after storage is formatted as described in [Generate and format IDs](#generate-format-ids).
- Type: string - Default: null - Importance: high ### Controller configuration example You can find the example KRaft configuration files in `/etc/kafka/`. You will see three different example files in this folder after you install Confluent Platform: - `broker.properties` - An example of the settings to use when the server is a broker only. - `controller.properties` - An example of the settings to use when the server is a controller only. - `server.properties` - An example of the settings to use when the server is both a broker and a controller. This configuration is not supported for production use. The following example shows an excerpt from a properties file for a controller on a system with three controllers. This example uses dynamic quorum, the recommended approach. ```bash ############################# Server Basics ############################# # The role of this server. Setting this puts us in KRaft mode. process.roles=controller # The node id associated with this instance's roles. node.id=1 # The bootstrap servers string for dynamic quorum. controller.quorum.bootstrap.servers=controller1.example.com:9093,controller2.example.com:9093,controller3.example.com:9093 ############################# Socket Server Settings ############################# # The address the socket server listens on. # Note that only the controller listeners are allowed here when `process.roles=controller`. # FORMAT: # listeners = listener_name://host_name:port # EXAMPLE: # listeners = PLAINTEXT://your.host.name:9092 listeners=CONTROLLER://controller1.example.com:9093 # A comma-separated list of the names of the listeners used by the controller. # This is required if running in KRaft mode. controller.listener.names=CONTROLLER # How to communicate with brokers. inter.broker.listener.name=BROKER # Maps listener names to security protocols, the default is for them to be the same. listener.security.protocol.map=CONTROLLER:SSL,BROKER:SSL ############################# Log Basics ############################# # A comma separated list of directories under which to store log files log.dirs=/tmp/kraft-controller-logs # ... # Additional property settings to match broker settings. ``` ### Security For configuring SASL/SCRAM for broker-to-broker communication, see [KRaft-based Confluent Platform clusters](../security/authentication/sasl/scram/overview.md#sasl-scram-kraft-based-clusters). For general security information for KRaft, see [KRaft Security in Confluent Platform](../security/component/kraft-security.md#kraft-security). If the `CONTROLLER` listener uses a security protocol other than `PLAINTEXT`, such as `SASL_PLAINTEXT`, `SASL_SSL`, or `SSL`, configure the broker to authenticate to the KRaft controller by setting the `sasl.mechanism.controller.protocol` property and the required `listener.name.controller..*` properties. The broker uses these settings to create the control-plane client that communicates with the controller. The following example shows a configuration where the `CONTROLLER` listener uses `SASL_PLAINTEXT` with the `PLAIN` mechanism: ```none listener.security.protocol.map=PLAINTEXT:PLAINTEXT,CONTROLLER:SASL_PLAINTEXT sasl.mechanism.controller.protocol=PLAIN listener.name.controller.plain.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \ username="" \ password=""; ``` ### Other properties KRaft controllers need a property only if: 1. The controller process itself uses that property to perform its controller duties, or 2. The property affects cluster metadata that controllers manage Brokers handle everything else—even if it’s cluster-wide configuration. If a property is only consumed by brokers for data-plane operations (like reading or writing data), it does not need to be in the controller’s configuration file. For example: - `confluent.schema.registry.url`: Controllers need this because Schema Registry integration affects metadata operations that controllers coordinate. Controllers must validate schema references during topic creation and modification, which is a metadata-layer function. - `confluent.tier.enable`: Controllers don’t need this because tiered storage is a broker-only, data-plane feature. Tiered storage operations (uploading and downloading segments to object storage) are only broker responsibilities. Based on this rule, many properties related to topic management and cluster-wide behavior must be present on the controllers because they affect cluster metadata that controllers manage. This includes security properties such as the truststore locations needed for secure communication with brokers. The following list provides an example of common settings that are required on KRaft controllers because they relate to cluster metadata. This is not an exhaustive list. - `auto.create.topics.enable` - `compression.type` - `confluent.metrics.reporter.bootstrap.servers` - `confluent.license.topic.replication.factor` - `confluent.metadata.topic.replication.factor` - `default.replication.factor` - `delete.topic.enable` - `message.max.bytes` - `metrics.reporters` - `min.insync.replicas` - `num.partitions` - `offsets.retention.minutes` - `offsets.topic.replication.factor` - `transaction.state.log.replication.factor` - `transaction.state.log.min.isr` - `unclean.leader.election.enable` ## Settings for other Kafka and Confluent Platform components You must use current, non-deprecated, configurations settings. The settings to use are described in the following table. | Feature | Allowed with ZooKeeper | Required with KRaft | |---------------------------|--------------------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | Clients and services | `zookeeper.connect=zookeeper:2181` | `bootstrap.servers=broker:9092` | | Schema Registry | `kafkastore.connection.url=zookeeper:2181` | `kafkastore.bootstrap.servers=broker:9092` | | Administrative tools | `kafka-topics --zookeeper zookeeper:2181` (deprecated) | `kafka-topics --bootstrap-server broker:9092` …
: `--command-config properties` to connect to brokers | | Retrieve Kafka cluster ID | `zookeeper-shell zookeeper:2181 get/cluster/id` | From the command line, use `kafka-metadata-quorum` (See [kafka-metadata-quorum](#describe-runtime-status))
or `confluent cluster describe --url`,
or view `metadata.properties`.
or `http://broker:8090 --output json` | ## Generate and format IDs Before you start Kafka, you must use the [kafka-storage](/kafka/operations-tools/kafka-tools.html#kafka-storage-sh) tool with the `random-uuid` command to generate a cluster ID for each new cluster. You only need one cluster ID, which you will use to format each node in the cluster. ```none bin/kafka-storage random-uuid ``` This results in output like the following: ```none q1Sh-9_ISia_zwGINzRvyQ ``` Then use the cluster ID to format storage for each node in the cluster with the `kafka-storage` tool that is provided with Confluent Platform, and the `format` command like the following example, specifying the properties file for a controller. ```none bin/kafka-storage format -t q1Sh-9_ISia_zwGINzRvyQ -c etc/kafka/controller.properties ``` Previously, Kafka would format blank storage directories automatically and generate a new cluster ID automatically. One reason for the change is that auto-formatting can sometimes obscure an error condition. This is particularly important for the metadata log maintained by the controller and broker servers. If a majority of the controllers were able to start with an empty log directory, a leader might be able to be elected with missing committed data. To configure the log directory, either set `metadata.log.dir` or `log.dirs`. For more information, see [Inter-broker listeners](#kraft-log-settings). ## Provision nodes Provision controller and broker nodes for KRaft clusters. Use dynamic controllers for new clusters or static controllers for fixed controller deployments. #### NOTE Use dynamic controller quorums for new clusters. Dynamic quorums let you add or remove controllers from a running cluster without updating configuration files on all nodes or performing full cluster restarts. Use `controller.quorum.bootstrap.servers` instead of `controller.quorum.voters` in your configuration to enable this operational flexibility. Create a cluster with multiple controllers using dynamic quorum. Starting with Confluent Platform 8.2, set `controller.quorum.auto.join.enable=true` on all controllers. Then, format the first controller with `--standalone`, and format each additional controller with `--no-initial-controllers`. For instructions, see [Provision multiple controllers for a new cluster](#provision-multiple-controllers). ### Provision multiple controllers for a new cluster Use one of the following options to provision multiple controllers for a new cluster. Both options produce a dynamic-quorum cluster. The difference is how the voter set is established. **Option 1: Bootstrap with –standalone, then rely on auto-join** Use this option when you want to roll out controllers sequentially. Starting with Confluent Platform 8.2, set `controller.quorum.auto.join.enable=true` so that additional controllers join the quorum automatically after a bootstrap controller is running. 1. Set `controller.quorum.auto.join.enable=true` in the configuration file for all controllers. 2. Generate a cluster ID: ```bash bin/kafka-storage.sh random-uuid ``` Save this cluster ID. Use it for all nodes in the cluster. 3. Format and start the first controller using the `--standalone` flag: ```bash bin/kafka-storage.sh format --cluster-id --standalone --config config/controller.properties bin/kafka-server-start.sh config/controller.properties ``` This bootstraps the first controller as the sole voter in the quorum. 4. Format and start each additional controller using the `--no-initial-controllers` flag: ```bash bin/kafka-storage.sh format --cluster-id --no-initial-controllers --config config/controller.properties bin/kafka-server-start.sh config/controller.properties ``` With `controller.quorum.auto.join.enable=true`, each controller automatically joins the quorum without any manual `add-controller` commands. **Option 2: Format every controller with –initial-controllers** Use this option when you want to define the voter set upfront and avoid the bootstrap step. Every controller is formatted with the same `--initial-controllers` argument, which uses the format `id@hostname:port:directoryId` and requires a directory UUID for each controller. 1. Generate a cluster ID: ```bash bin/kafka-storage.sh random-uuid ``` Save this cluster ID. Use it for all nodes in the cluster. 2. Generate a directory UUID for each controller. The `--initial-controllers` argument requires a directory UUID for every controller in the voter set. Use `kafka-storage.sh random-uuid` to produce one UUID per controller, and record the value alongside the controller’s node ID, hostname, and port. You use these values in the next step. ```bash bin/kafka-storage.sh random-uuid ``` Record each UUID alongside its controller’s node ID, hostname, and port. For a three-node controller quorum, your notes look similar to the following: ```text 0@controller-0:9093:JEXY6aqzQY-32P5TStzaFg 1@controller-1:9093:MvDxzVmcRsaTz33bUuRU6A 2@controller-2:9093:07R5amHmR32VDA6jHkGbTA ``` Combine the per-controller values into a single comma-separated string for the `--initial-controllers` argument. For example, given controller IDs `0`, `1`, and `2`; hostnames `controller-0`, `controller-1`, and `controller-2`; port `9093`; and directory IDs `A`, `B`, and `C`, the value of `--initial-controllers` is: ```text 0@controller-0:9093:A,1@controller-1:9093:B,2@controller-2:9093:C ``` 3. On each controller node, format the storage with the `--initial-controllers` flag: ```bash bin/kafka-storage.sh format --cluster-id \ --initial-controllers \ --config config/controller.properties ``` Replace the following values: - ``: The cluster ID from step 1. - ``: The comma-separated list of controller node IDs, hostnames, ports, and directory UUIDs from step 2, formatted as `id@hostname:port:directoryId`. #### IMPORTANT - The `--initial-controllers` value must be identical on all controller nodes. The same node IDs, hostnames, ports, and directory UUIDs must appear in the same order on every invocation. - Each controller must use the same cluster ID. - The node IDs, hostnames, and ports in the list must match the configuration in each controller’s `controller.properties` file. 4. Start each controller: ```bash bin/kafka-server-start.sh config/controller.properties ``` ### Provision brokers and add new controllers After your initial controller quorum is running, format brokers and any additional controllers using the `--no-initial-controllers` flag. This tells the node to join an existing quorum rather than create a new one. To format broker storage, run the following command: ```bash bin/kafka-storage.sh format --cluster-id --no-initial-controllers --config config/server.properties ``` To format storage for new controllers that join an existing dynamic quorum, set `controller.quorum.auto.join.enable=true` in the controller’s configuration file, then run the following command: ```bash bin/kafka-storage.sh format --cluster-id --no-initial-controllers --config config/controller.properties ``` After you format the storage, start the controller. On Confluent Platform 8.2 and later, the controller automatically joins the quorum. On earlier versions, add the new controller to the quorum using the `kafka-metadata-quorum` tool. For more details, see [Controller membership changes](#controller-membership-changes). ### Configuration requirements All brokers and controllers in the cluster must include `controller.quorum.bootstrap.servers` in their configuration files. Do not include `controller.quorum.voters` when using dynamic controllers. The `controller.quorum.bootstrap.servers` property lets nodes discover the quorum: ```properties controller.quorum.bootstrap.servers=controller1.example.com:9093,controller2.example.com:9093,controller3.example.com:9093 ``` Unlike `controller.quorum.voters`, the `controller.quorum.bootstrap.servers` property does not need to list every controller. List enough controllers to let nodes locate the quorum. However, include as many controller endpoints as feasible for reliability. For complete controller and broker configuration examples, see [Required settings](#server-basics). ### Configure SCRAM To configure SCRAM for brokers in a Kafka cluster running in KRaft mode, you must create the credentials before your brokers are up and running. You then use the `--add-scram` option with the `kafka-storage` tool. For more information, see [SASL for KRaft-based clusters](../security/authentication/sasl/scram/overview.md#sasl-scram-kraft-based-clusters). ## Upgrade from static controllers to dynamic controllers Confluent Platform 8.0 and later supports upgrading a KRaft cluster from a static controller configuration to a dynamic controller configuration. Dynamic controllers provide significantly greater operational flexibility. Unlike the static approach, a dynamic configuration allows you to add or remove controllers from a running cluster without updating configuration files on all nodes or performing a full cluster restart. See [KRaft server compatibility](../installation/versions-interoperability.md#kraft-server-compatibility) for compatibility information. ### Static versus dynamic quorums Understanding the difference in configuration is key to the upgrade process. + **Static quorum (original approach)**: Requires the `controller.quorum.voters` property on every broker and controller. This list must explicitly specify the IDs, hostnames, and ports of all controllers. You cannot modify the quorum without updating this configuration on every node and restarting them. + **Dynamic quorum (recommended)**: Uses the `controller.quorum.bootstrap.servers` property instead. Much like the `bootstrap.servers` configuration used by Kafka clients, this key does not need to list every controller—only enough to allow servers to locate the quorum. This enables you to change the controller set dynamically while the cluster is running. ### Prerequisites Before starting the upgrade, ensure that: + Your cluster is running Confluent Platform 8.0 or later. + All brokers and controllers are running and healthy. + You have a backup of your cluster configuration. ### Verify current KRaft version Dynamic controller clusters are added in KRaft version `1` (Kafka 4.0). To determine which KRaft feature version your cluster is using, execute the following CLI command: ```bash bin/kafka-features.sh --bootstrap-controller localhost:9093 describe ``` If the `kraft.version` field shows `FinalizedVersionLevel: 0` or is absent, you are using a static quorum. If it shows `FinalizedVersionLevel: 1` or above, you are using a dynamic quorum. Example output for a static quorum: ```none Feature: kraft.version SupportedMinVersion: 0 SupportedMaxVersion: 1 FinalizedVersionLevel: 0 Epoch: 9 ``` Example output for a dynamic quorum: ```none Feature: kraft.version SupportedMinVersion: 0 SupportedMaxVersion: 1 FinalizedVersionLevel: 1 Epoch: 9 ``` ### Upgrade KRaft version If your `FinalizedVersionLevel` is `0`, you must upgrade the feature level before changing configurations If `FinalizedVersionLevel` is `1`, you can skip to [Update KRaft configuration](#update-kraft-config). To upgrade the KRaft version from `0` to `1`, use the `kafka-features` tool: ```bash bin/kafka-features.sh --bootstrap-controller localhost:9093 upgrade --feature kraft.version=1 ``` After the upgrade completes, verify the new version: ```bash bin/kafka-features.sh --bootstrap-controller localhost:9093 describe ``` The output should show `FinalizedVersionLevel: 1` for `kraft.version`: ```none Feature: kraft.version SupportedMinVersion: 0 SupportedMaxVersion: 1 FinalizedVersionLevel: 1 Epoch: 8 ``` ### Update KRaft configuration After upgrading the KRaft version, you must update the configuration on all brokers and controllers to use `controller.quorum.bootstrap.servers` instead of `controller.quorum.voters`. 1. Update controller configuration on each controller node: 1. Remove or comment out the `controller.quorum.voters` property. 2. Add the `controller.quorum.bootstrap.servers` property with a comma-separated list of controller endpoints. The `controller.quorum.bootstrap.servers` property should contain as many controller endpoints as possible so that all servers can locate the quorum. Unlike `controller.quorum.voters`, this property does not need to contain all controllers, but it should contain enough to ensure connectivity. Example `server.properties` file configuration: ```bash # Remove or comment out: # controller.quorum.voters=1@controller1.example.com:9093,2@controller2.example.com:9093,3@controller3.example.com:9093 # Add: controller.quorum.bootstrap.servers=controller1.example.com:9093,controller2.example.com:9093,controller3.example.com:9093 ``` 2. Update broker configuration on each broker node: 1. Remove or comment out the `controller.quorum.voters` property. 2. Add the `controller.quorum.bootstrap.servers` property with the same list of controller endpoints. 3. Restart all nodes: 1. Restart each controller node, one at a time, allowing each to fully start before restarting the next. 2. After all controllers are restarted and healthy, restart each broker, one at a time. 4. Verify the upgrade by running the following command: ```bash bin/kafka-metadata-quorum.sh --bootstrap-server localhost:9092 describe --status ``` The cluster should show the same number of controllers as before, and all should be listed as `CurrentVoters`. ```bash ClusterId: M51W2LQQSRiDQ1tA0l8tew LeaderId: 1 LeaderEpoch: 1 HighWatermark: 148595 MaxFollowerLag: 0 MaxFollowerLagTimeMs: 0 CurrentVoters: [{"id": 1, "directoryId": "QWOv0BvKQP2KwR_5kMXYUA", "endpoints": ["CONTROLLER://localhost:9093"]}] CurrentObservers: [{"id": 2, "directoryId": "R5VtfFv_hW-tlqYssdW7iw"}] ``` ## Controller membership changes After upgrading to dynamic controllers, you can add or remove controllers from your cluster without updating configuration files on all nodes or performing a full cluster restart. ### Add a new controller If a dynamic controller cluster already exists, it can be expanded by first provisioning a new controller using the `kafka-storage.sh` tool and starting the controller. After starting the controller, the replication to the new controller can be monitored using the `bin/kafka-metadata-quorum.sh describe --replication` command. Once the new controller has caught up to the active controller, it can be added to the cluster using the `bin/kafka-metadata-quorum.sh add-controller` command. To add a new controller: 1. Provision the new controller node with the appropriate configuration. For details about required configuration properties, see [Required settings](#server-basics). 2. Format the storage directory for the new controller using the `kafka-storage` tool. For information about formatting storage, including the `--no-initial-controllers` flag, see [Generate and format IDs](#generate-format-ids). 3. Start the new controller and allow it to start up. 4. Monitor the replication status to ensure the new controller catches up to the active controller: When using broker endpoints: ```bash bin/kafka-metadata-quorum.sh --bootstrap-server localhost:9092 describe --replication ``` When using controller endpoints: ```bash bin/kafka-metadata-quorum.sh --bootstrap-controller localhost:9093 describe --replication ``` 5. Once the new controller has caught up to the active controller, add it to the cluster using the `kafka-metadata-quorum` tool. When using broker endpoints: ```bash bin/kafka-metadata-quorum.sh --command-config config/controller.properties --bootstrap-server localhost:9092 add-controller ``` When using controller endpoints: ```bash bin/kafka-metadata-quorum.sh --command-config config/controller.properties --bootstrap-controller localhost:9093 add-controller ``` ### Remove a controller If the dynamic controller cluster already exists, it can be shrunk using the `bin/kafka-metadata-quorum.sh remove-controller` command. Run the remove-controller command before shutting down the controller to remove it from the quorum first. You can obtain the controller ID and directory ID from the `kafka-metadata-quorum.sh describe --status` command output. For more information, see [Describe runtime status](#describe-runtime-status). When using broker endpoints: ```bash bin/kafka-metadata-quorum.sh --bootstrap-server localhost:9092 remove-controller --controller-id --controller-directory-id ``` When using controller endpoints: ```bash bin/kafka-metadata-quorum.sh --bootstrap-controller localhost:9093 remove-controller --controller-id --controller-directory-id ``` ## Tools for debugging KRaft mode Kafka provides tools to help you debug a cluster running in KRaft-mode. ### Describe runtime status You can describe the runtime state of the cluster metadata partition using the [kafka-metadata-quorum tool](/kafka/operations-tools/kafka-tools.html#kafka-metadata-quorum-sh) and specify either a Kafka broker with the `--bootstrap-server` option or a KRaft controller with the `--bootstrap-controller` option. For example, the following command specifies a broker and displays a summary of the metadata quorum: ```none bin/kafka-metadata-quorum --bootstrap-server host1:9092 describe --status ``` ```none Output might look like the following: ClusterId: fMCL8kv1SWm87L_Md-I2hg LeaderId: 3002 LeaderEpoch: 2 HighWatermark: 10 MaxFollowerLag: 0 MaxFollowerLagTimeMs: -1 CurrentVoters: [3000,3001,3002] CurrentObservers: [0,1,2] ``` You can specify a controller with the `--bootstrap-controller` option. This is useful when the brokers are not accessible. ```none bin/kafka-metadata-quorum --bootstrap-controller host1:9093 describe --status ``` ### Debug log segments The [kafka-dump-log tool](/kafka/operations-tools/kafka-tools.html#kafka-dump-log-sh) tool can be used to debug the log segments and snapshots for the cluster metadata directory. The tool will scan the provided files and decode the metadata records. For example, the following command decodes and prints the records in the first log segment: ```none bin/kafka-dump-log --cluster-metadata-decoder --files tmp/kraft-controller-logs/_cluster_metadata-0/00000000000000023946.log ``` ### Inspect the metadata partition You can use the `kafka-metadata-shell` to inspect the metadata partition. The Kafka version of the `kafka-metadata-shell` tool enables you to interactively examine the metadata stored in a KRaft cluster. To analyze metadata, point the `kafka-metadata-shell` tool to a log directory, run the `kafka-metadata-shell.sh` command, using the `--directory` flag to specify the path to your cluster metadata log. ```bash ./kafka-metadata-shell.sh --directory /tmp/kraft-combined-logs/__cluster_metadata-0/ ``` Once the shell loads, you can use commands like `ls` and `cat` to explore the metadata records. For more information, see [kafka-metadata-shell tool](/kafka/operations-tools/kafka-tools.html#kafka-metadata-shell.sh). The Confluent Platform version of `kafka-metadata-shell` includes additional options for connecting directly to a running controller, which is necessary for clusters using security features like SSL/TLS. The following is the help output for the `kafka-metadata-shell` tool. ```bash kafka-metadata-shell.sh --help usage: kafka-metadata-shell.sh [-h] [--cluster-id CLUSTER_ID] [--offset OFFSET] [--config CONFIG] (--directory DIRECTORY | --controllers CONTROLLERS) [command [command ...]] The Apache Kafka metadata shell tool positional arguments: command The command to run. optional arguments: -h, --help show this help message and exit --cluster-id CLUSTER_ID, -t CLUSTER_ID The cluster id. Required when using --controllers --directory DIRECTORY, -d DIRECTORY The __cluster_metadata-0 directory to read. --controllers CONTROLLERS, -q CONTROLLERS The controller.quorum.voters. --offset OFFSET, -o OFFSET The (exclusive) offset to read up to --config CONFIG Path to property file containing a Kafka configuration ``` You use the `--config` flag to point to a client properties file that contains your security settings. **Example: Connecting to a controller over SSL** The following example shows how to connect to a controller that’s configured to require SSL on its listener. 1. Create a client configuration file. This file contains the properties the shell needs to authenticate with the controller. ```bash # client.properties # Maps the listener name from the server to the SSL security protocol. listener.security.protocol.map=CONTROLLER:SSL,PLAINTEXT:PLAINTEXT # Provides the location and password for your client's truststore. ssl.truststore.location=/path/to/your/truststore.jks ssl.truststore.password=your-password ssl.truststore.type=JKS ``` 1. Run the `kafka-metadata-shell` command, pointing to your controller, cluster ID, and the new client configuration file: ```bash ./bin/kafka-metadata-shell \ --cluster-id your-cluster-id \ --controllers controller-host:9593 \ --config /path/to/your/client.properties ``` The shell uses the properties in the config file to establish a secure, trusted connection to the controller. ### Check migration status You can check the migration status of a KRaft cluster using Confluent-provided `kafka-migration-check` tool. This tool is included in Confluent Platform 7.9.2 or later, and can be found in the `bin` directory of your Confluent Platform installation. For detailed instructions on how to run this tool, see [Check Clusters for KRaft Migration](../tools/kraft-migration-tool.md#kraft-migration-tool). ## Monitor KRaft Following are some JMX metrics to monitor on the controller and broker when operating in KRaft mode. Some of the metrics depend on the setting for [process.roles](#server-basics). For more broker metrics, see [Broker metrics](../kafka/broker-metrics.md#kafka-monitoring-metrics-broker). ### KRaft quorum monitoring metrics The following table lists KRaft controller quorum metrics. | `kafka.server:type=raft-metrics` MBean name | Description | |-----------------------------------------------|----------------------------------------------------------------------------------------------------------------------| | `append-records-rate` | The average number of records appended per second by the leader of the raft quorum. | | `commit-latency-avg` | The average time in milliseconds to commit an entry in the raft log. | | `commit-latency-max` | The maximum time in milliseconds to commit an entry in the raft log. | | `current-epoch` | The current quorum epoch. | | `current-leader` | The current quorum leader’s id; -1 indicates unknown. | | `current-state` | The current state of this member; possible values are leader, candidate, voted, follower, unattached, observer. | | `current-vote` | The current voted leader’s id; -1 indicates not voted for anyone. | | `election-latency-avg` | The average time in milliseconds spent on electing a new leader. | | `election-latency-max` | The maximum time in milliseconds spent on electing a new leader. | | `fetch-records-rate` | The average number of records fetched from the leader of the raft quorum. | | `high-watermark` | The high watermark maintained on this member; -1 if it is unknown. | | `log-end-offset` | The current raft log end offset. | | `number-unknown-voter-connections` | Number of unknown voters whose connection information is not cached. This value of this metric is always 0. | | `poll-idle-ratio-avg` | The average fraction of time the client’s poll() is idle as opposed to waiting for the user code to process records. | Other quorum metrics: | MBean | Description | |-----------------------------------------------------------------------|--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | `kafka.server:type=MetadataLoader,name=CurrentMetadataVersion` | Outputs the feature level of the current metadata version. | | `kafka.server:type=MetadataLoader,name=HandleLoadSnapshotCount` | The total number of times that a KRaft snapshot has been loaded since the process was started. | | `kafka.server:type=SnapshotEmitter,name=LatestSnapshotGeneratedBytes` | The total size in bytes of the latest snapshot that the node has generated.
If a snapshot has not been generated yet, this is the size of the latest snapshot that was loaded.
If no snapshots have been generated or loaded, this is 0. | | `kafka.server:type=SnapshotEmitter,name=LatestSnapshotGeneratedAgeMs` | The interval in milliseconds since the latest snapshot was generated.
If no snapshot has been generated yet, this is the approximate time delta since the process was started. | ### Controller metrics With KRaft, Kafka adds a controller quorum to the cluster. These controllers must be able to commit records for Kafka to be available so you need to monitor their health. For the full list of KRaft metrics, see [KRaft broker metrics](../kafka/broker-metrics.md#kraft-broker-metrics) and [KRaft Quorum metrics](../kafka/broker-metrics.md#kraft-quorum-metrics). | `kafka.controller:type=KafkaController` MBean name | Description | |------------------------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | `ActiveBrokerCount` | When using KRaft, the number of registered and unfenced brokers as observed by this controller. When using ZooKeeper, this value is the number of brokers known to the controller. | | `ActiveControllerCount` | The number of active controllers on this node. Valid values are ‘0’ or ‘1’.
Alert if the aggregated sum across all brokers in the cluster is anything other than 1 because there should be exactly one controller per cluster. | | `FencedBrokerCount` | When using KRaft, the number of registered but fenced brokers as observed by this controller. | | `GlobalPartitionCount` | The number of all partitions in the cluster as observed by this controller. | | `GlobalTopicCount` | The number of all topics in the cluster as observed by this controller. | | `LastAppliedRecordLagMs` | Reports the difference between the local time and the append time of the last applied record batch.
For active controllers the value of this lag is always zero. | | `LastAppliedRecordOffset` | The offset of the last record that was applied by the controller to the cluster metadata partition.
For the active controller this may include uncommitted records. For the inactive controller this always includes committed records only. | | `LastAppliedRecordTimestamp` | The timestamp of the last record that was applied by the controller to the cluster metadata partition. | | `LastCommittedRecordOffset` | The active controller reports the offset of the last committed offset it consumed.
Inactive controllers will always report the same value as `LastAppliedRecordOffset`. You can monitor the last committed offsets to see that they are advancing.
You can also use these metrics to check that all of the brokers and controllers are at a similar offset. | | `LastAppliedRecordTimestamp` | The timestamp of the last record that was applied by the controller to the cluster metadata partition. | | `MetadataErrorCount` | The number of times this controller node has encountered an error during metadata log processing. | | `NewActiveControllerCount` | Counts the number of times this node has seen a new controller elected.
A transition to the “no leader” state is not counted here. If the same controller as before becomes active, that still counts. | | `EventQueueOperationsStartedCount` | The total number of controller event queue operations that were started.
This count includes deferred operations. | | `EventQueueOperationsTimedOutCount` | The total number of controller event queue operations that timed out before they could be performed. | | `OfflinePartitionsCount` | The number of offline topic partitions (non-internal) as observed by this controller. | | `PreferredReplicaImbalanceCount` | The count of topic partitions for which the leader is not the preferred leader. | | `TimedOutBrokerHeartbeatCount` | The number of broker heartbeats that timed out on this controller since the process was started.
Note that only active controllers handle heartbeats, so only they will see increases in this metric. | ControllerEventManager metrics: | `kafka.controller:type=ControllerEventManager` MBean name | Description | |-------------------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------| | `AvgIdleRatio` | The fraction of time the controller event manager thread spends idle, where 0.0 means the thread is always busy and 1.0 means the thread is always idle. | | `EventQueueProcessingTimeMs` | A histogram of the time in milliseconds that requests spent being processed in the controller event queue. | | `EventQueueTimeMs` | A histogram of the time in milliseconds that requests spent waiting in the controller event queue. | ### KRaft Broker metrics | `kafka.server:type=broker-metadata-metrics` MBean name | Description | |----------------------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------| | `last-applied-record-offset` | The offset of the last record from the cluster metadata partition that was applied by the broker. | | `last-applied-record-timestamp` | The timestamp of the last record from the cluster metadata partition that was applied by the broker. | | `last-applied-record-lag-ms` | The difference between now and the timestamp of the last record from the cluster
metadata partition that was applied by the broker. | | `metadata-load-error-count` | The number of errors encountered by the `BrokerMetadataListener` while loading the
metadata log and generating a new metadata delta based on it. | | `metadata-apply-error-count` | The number of errors encountered by the `BrokerMetadataPublisher`
while applying a new metadata imaged based on the latest metadata delta. | ## Related content - [KRaft Overview for Confluent Platform](kraft.md#kraft-overview) - [Quick Start for Confluent Platform](../get-started/platform-quickstart.md#quickstart) - [ZooKeeper Topic Guide](zk-production.md#zk-prod-deployment)