Using KRaft Consensus Protocol
1. Understanding KRaft Architecture
| Component | Description | Role |
|---|---|---|
| Controller | controller |
Manages cluster state and assigns partitions to brokers |
| Broker | broker |
Handles data storage and client requests |
| Metadata Log | metadata.log |
Stores cluster metadata and state changes |
Example: KRaft Controller Configuration
controller.quorum.voters=127.0.0.1:9093,127.0.0.1:9094,127.0.0.1:9095
Note: KRaft is the default mode for Kafka 3.0+ and removes the dependency on ZooKeeper.
KRaft Architecture Overview
- Controller manages cluster state and partition assignments
- Brokers store and serve data
- Metadata is stored in a Raft log for consistency
2. Understanding Metadata Replication
| Aspect | Description | Details |
|---|---|---|
| Metadata Storage | Location of metadata | Stored in Kafka's internal topics like __cluster_metadata or __raft_metadata |
| Replication Factor | Number of replicas | Default is 3; can be configured per topic or cluster-wide |
| Leader Election | Process of selecting a leader | Based on Raft consensus protocol; ensures consistency and availability |
| Quorum Writes | Write consistency requirement | Writes must be acknowledged by a majority of replicas before being considered committed |
Example: Configuring Metadata Replication
{<br> "metadata.replication.factor": 3,<br> "min.insync.replicas": 2<br>}
Note: Metadata replication ensures fault tolerance and consistency in KRaft mode.
3. Configuring Metadata Log
| Property | Default Value | Description |
|---|---|---|
| log.dirs | /tmp/kafka-logs |
Directories where Kafka stores log segments. Multiple paths can be specified, separated by commas. |
| log.flush.interval.messages | 10000 |
Number of messages to accumulate before flushing the log to disk. |
| log.flush.interval.ms | 900000 |
Time in milliseconds to wait before flushing the log to disk, regardless of message count. |
| log.retention.hours | 168 |
Retention period for logs in hours. After this time, older log segments are deleted. |
| log.retention.bytes | -1 |
Maximum size of the log in bytes. When exceeded, older segments are deleted. |
| log.segment.bytes | 1073741824 |
Maximum size of a single log segment in bytes. |
| log.roll.hours | 168 |
Time in hours after which a new log segment is rolled, even if it's not full. |
Example: Configuring Metadata Log
log.dirs=/var/lib/kafka/data
log.flush.interval.messages=10000
log.flush.interval.ms=900000
log.retention.hours=168
log.segment.bytes=1073741824
log.roll.hours=1684. Setting Controller Quorum Voters
| Property | Default Value | Description |
|---|---|---|
| controller.quorum.voters | Empty list | List of controller quorum voters in the format <id>@<host>:<port> |
Example: Configuring Controller Quorum Voters
{
"controller.quorum.voters": "0@localhost:9093,1@localhost:9094,2@localhost:9095"
}Note: The
controller.quorum.voters property must be set on all brokers in the cluster to ensure consistent quorum configuration.Warning: Changing the controller quorum voters after cluster initialization can lead to loss of quorum and cluster unavailability.
5. Formatting Storage
Example: Formatting Kafka Storage Directories
bin/kafka-storage.sh format -c config/server.properties -t <cluster-id>| Option | Values | Description |
|---|---|---|
| -c | config/server.properties |
Path to the server configuration file |
| -t | <cluster-id> |
Unique identifier for the Kafka cluster |
| --ignore-sanity-checks | false |
Disable sanity checks before formatting (not recommended) |
Note: The
cluster-id must be unique per Kafka cluster and should be generated using a UUID generator.
Warning: Formatting storage will erase all existing data in the Kafka log directories. Ensure backups are taken before running this command.
6. Starting KRaft Controller
To start a Kafka cluster using the KRaft consensus protocol, you need to configure the broker properties and initialize the controller quorum.
| Property | Value | Description |
|---|---|---|
| node.id | 0 |
Unique identifier for the broker node. |
| process.roles | controller |
Specifies that this node acts as a controller in KRaft mode. |
| controller.quorum.voters | 0@localhost:9093 |
List of controller voters in the format nodeId@host:port. |
| listeners | PLAINTEXT://localhost:9092 |
Listener for client connections. |
| controller.listener.names | PLAINTEXT |
Name of the listener used for inter-broker communication. |
Example: Starting KRaft Controller
# Start the Kafka broker in KRaft mode
bin/kafka-server-start.sh config/server.properties
Note: Ensure that the
process.roles property is set to controller for the controller node.
Warning: The KRaft mode is still experimental and not recommended for production use.
7. Starting KRaft Broker
To start a Kafka broker using the KRaft consensus protocol, you must configure the broker to run in KRaft mode and provide the necessary metadata directory.
| Configuration Property | Value | Description |
|---|---|---|
| process.roles | broker |
Specifies that this node is a broker in KRaft mode. |
| node.id | 1 |
Unique identifier for the broker in the cluster. |
| listeners | PLAINTEXT://localhost:9092 |
Network listeners for client connections. |
| inter.broker.listener.name | PLAINTEXT |
Listener name used for inter-broker communication. |
| metadata.log.dir | /tmp/kafka-logs/metadata |
Directory where metadata logs are stored. |
Note: Ensure that the metadata log directory is writable and has sufficient space for logs.
Warning: Do not run KRaft in production without proper configuration and testing.
8. Using Combined Mode
Kafka's Combined Mode is a configuration that allows you to run Kafka in a single process, combining both the controller and broker roles. This mode is particularly useful for development and testing environments where you want to simplify the setup.
| Configuration | Description | Default Value | Notes |
|---|---|---|---|
| process.roles | Defines the roles that this Kafka process will handle. | broker |
Set to broker or controller in standalone mode, or broker,controller for combined mode. |
| node.id | Unique identifier for the Kafka node. | -1 |
Must be unique across all nodes in the cluster. |
| controller.quorum.voters | List of voters in the controller quorum. | "" |
Required in KRaft mode; must include all controller nodes. |
Example: Combined Mode Configuration
# Kafka server configuration for Combined Mode
process.roles=broker,controller
node.id=1
controller.quorum.voters=1@localhost:9093
listeners=PLAINTEXT://localhost:9092
advertised.listeners=PLAINTEXT://localhost:9092
log.dirs=/tmp/kafka-logs
Note: Combined Mode is not recommended for production use due to its single-process nature, which can lead to performance bottlenecks and reduced fault tolerance.
9. Configuring Controller Listener
| Property | Default Value | Description |
|---|---|---|
| controller.listener.names | |
Comma-separated list of listener names used by the controller. Must be set when using KRaft. |
| listener.security.protocol.map | PLAINTEXT:PLAINTEXT,SSL:SSL,SASL_PLAINTEXT:SASL_PLAINTEXT,SASL_SSL:SASL_SSL |
Maps listener names to security protocols. Required for KRaft. |
| controller.quorum.voter.id | 0 |
Unique identifier for the controller node in the quorum. Must be unique per node. |
Example: Controller Listener Configuration
controller.listener.names=PLAINTEXT,SSL
listener.security.protocol.map=PLAINTEXT:PLAINTEXT,SSL:SSL
advertised.listeners=PLAINTEXT://kafka1.example.com:9092,SSL://kafka1.example.com:9093
Note: When using KRaft, the controller must be able to communicate with other nodes via the configured controller listeners. Ensure that the
controller.listener.names and listener.security.protocol.map are correctly set to allow secure communication.
10. Migrating from ZooKeeper
Note: KRaft is the new consensus protocol for Apache Kafka, designed to eliminate the dependency on ZooKeeper. Migration from ZooKeeper-based Kafka clusters to KRaft is a major operational task that requires careful planning and execution.
| Migration Step | Description | Notes |
|---|---|---|
| Enable KRaft mode | process.roles=broker and controller.quorum.voters=... |
Must be set in server.properties for all brokers |
| Configure Controller Quorum | Define controller nodes using controller.quorum.voters |
Each voter must have a unique ID and host:port address |
| Transition Metadata | Use kafka-storage.sh to format storage with KRaft metadata |
Requires stopping all brokers and reformatting disks |
| Start Brokers in KRaft Mode | Start all brokers with KRaft mode enabled | Brokers will form a new controller quorum and begin operation |
Example: Enabling KRaft in server.properties
process.roles=broker<br>controller.quorum.voters=1@localhost:9093,2@localhost:9094,3@localhost:9095<br>node.id=1<br>listeners=PLAINTEXT://localhost:9092<br>advertised.listeners=PLAINTEXT://localhost:9092
Migration Steps Summary
- Prepare all brokers to run in KRaft mode by setting
process.roles=broker - Define the controller quorum voters in
controller.quorum.voters - Format storage using
kafka-storage.shto initialize KRaft metadata - Start all brokers and verify quorum formation
Warning: Migration from ZooKeeper to KRaft is a destructive operation. Existing metadata in ZooKeeper will not be preserved and must be migrated manually or recreated.