Chapter 21. Migrating ZooKeeper-based Kafka clusters

Procedure

1. Verify that ZooKeeper is running.

   jcmd | grep zookeeper

   Returns:

   <process_id> org.apache.zookeeper.server.quorum.QuorumPeerMain ./config/zookeeper.properties

2. Check that the Kafka brokers are running:

   jcmd | grep kafka

   Returns:

   <process_id> kafka.Kafka ./config/server.properties

   Each Kafka broker appears as a running process.

3. Verify that each broker is connected to ZooKeeper.

   Use the ZooKeeper shell to list the registered brokers:

   ./bin/zookeeper-shell.sh <zk-host>:2181 ls /brokers/ids

   Returns a list of the IDs for the brokers currently registered with ZooKeeper. This command confirms the brokers are registered with ZooKeeper. Client connectivity is verified in the next step.

4. Check the cluster is running correctly by sending and consuming messages.

   For more information, see Sending and receiving messages from a topic.

Procedure

1. Retrieve the Kafka cluster ID using the zookeeper-shell tool:

   ./bin/zookeeper-shell.sh localhost:2181 get /cluster/id

   The command prints connection information before returning the cluster ID. The final JSON line in the output contains the cluster ID. For example:

   Connecting to localhost:2181
   
   WATCHER::
   WatchedEvent state:SyncConnected type:None path:null
   {"version":"1","id":"SY9T31EKSS6XeNO_Jio6yQ"}

   In this example, the cluster ID is SY9T31EKSS6XeNO_Jio6yQ.

2. Install a KRaft controller quorum to the cluster.

   A KRaft controller quorum is required to manage metadata during the migration. Each controller runs in a dedicated role and must be configured separately.

   1. Create a controller.properties file for each controller node.

      Each controller requires:

      • A unique node.id. The node.id must be unique across the entire Kafka cluster, including the existing brokers (which still use broker.id). Brokers and controllers share a single ID namespace, so do not reuse an ID already assigned to a broker or another controller.
      • The migration flag set to true.
      • ZooKeeper connection details.
      • A listener for controller-quorum communication.
      • A list of all controllers in the quorum.
      • A listener name for inter-broker communication.

      • A dedicated controller listener port that does not conflict with broker listener ports on the same host.

        Note

        The controller listener is used for communication within the KRaft quorum and for controller-related requests from brokers and administrative clients. It is not used for regular client traffic such as producing or consuming messages.

        Example controller configuration

        process.roles=controller
        node.id=1
        
        # Enable KRaft metadata migration
        zookeeper.metadata.migration.enable=true
        zookeeper.connect=zoo1.example.com:2181,zoo2.example.com:2181,zoo3.example.com:2181
        
        # Listener for quorum communication
        listeners=CONTROLLER://0.0.0.0:9090
        controller.listener.names=CONTROLLER
        listener.security.protocol.map=CONTROLLER:PLAINTEXT
        
        # List all controllers in the quorum
        controller.quorum.bootstrap.servers=controller1:9090,controller2:9090,controller3:9090
        
        # Required for migration
        inter.broker.listener.name=PLAINTEXT

        Use a dedicated port for the controller listener that does not conflict with broker listeners on each host. In most deployments, the same controller listener port is used on every controller node, and the host name distinguishes each controller. Only controllers configure the quorum voters.

   2. Generate a directory ID for each controller node.

      CONTROLLER_1_DIR_ID="$(./bin/kafka-storage.sh random-uuid)"
      CONTROLLER_2_DIR_ID="$(./bin/kafka-storage.sh random-uuid)"
      CONTROLLER_3_DIR_ID="$(./bin/kafka-storage.sh random-uuid)"

      Each controller requires a unique directory ID for metadata storage initialization.

   3. Format the metadata directory on each controller node.

      Use the cluster ID retrieved from ZooKeeper and specify the initial controller quorum:

      ./bin/kafka-storage.sh format \
        --cluster-id <cluster-id> \
        --initial-controllers "1@controller1:9090:${CONTROLLER_1_DIR_ID},2@controller2:9090:${CONTROLLER_2_DIR_ID},3@controller3:9090:${CONTROLLER_3_DIR_ID}" \
        --config ./config/kraft/controller.properties

      Use the same cluster ID and --initial-controllers configuration on all controllers. Each controller entry uses the format <node.id>@<host>:<port>:<directory.id>.

      By default, log.dirs is /tmp/kraft-controller-logs, which is cleared on system reboot. Set a persistent directory in production environments.

   4. Start each controller node.

      ./bin/kafka-server-start.sh -daemon ./config/kraft/controller.properties

   5. Verify that the controllers are running.

      jcmd | grep kafka

      You should see one process per controller.

   6. Confirm that each controller has entered the PRE_MIGRATION state.

      Run the following command on each controller host:

      grep "NONE to PRE_MIGRATION" ./logs/controller.log

      A log entry similar to the following indicates that the controller has started successfully, formed a quorum, and is ready for brokers to register:

      Replayed a ZkMigrationStateRecord changing the migration state from NONE to PRE_MIGRATION

      This state confirms that the controllers are prepared to begin metadata migration when all brokers have been restarted with migration enabled.

3. Enable migration on each broker.

   In this step, update each broker so it can register with the KRaft controller quorum and enter the migration process. Metadata migration does not begin immediately. The controllers start migrating metadata from ZooKeeper to KRaft only after all brokers have been restarted with migration enabled and have successfully registered with the controller quorum. During this phase, any new metadata created through brokers is written to KRaft first and then copied back to ZooKeeper by the controllers (dual-write) to maintain rollback compatibility. Controllers must be running before you restart any brokers.

   1. Stop the broker running on the host.

      ./bin/kafka-server-stop.sh
      jcmd | grep kafka

      If you are using a multi-node cluster, restart brokers one at a time. For guidance, see Section 3.8, “Performing a graceful rolling restart of Kafka brokers”.

   2. Update the broker configuration.

      Edit the server.properties file to enable metadata migration and to add a listener for communication with the KRaft controllers.

      Each broker requires:

      • inter.broker.protocol.version set to a supported version
      • The migration flag set to true
      • ZooKeeper connection details (already present on existing brokers)
      • A controller listener on a free port

      • Controller quorum connection details (must list all controllers)

        Example broker configuration

        broker.id=0
        
        # Set the inter-broker protocol version for migration
        inter.broker.protocol.version=3.9
        
        # Enable dual-write migration mode
        zookeeper.metadata.migration.enable=true
        zookeeper.connect=zoo1.my-domain.com:2181,zoo2.my-domain.com:2181,zoo3.my-domain.com:2181
        
        # Add a controller listener for KRaft quorum communication
        listeners=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
        listener.security.protocol.map=PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT
        
        # Connect the broker to the controller quorum
        controller.listener.names=CONTROLLER
        controller.quorum.bootstrap.servers=controller1:9090,controller2:9090,controller3:9090

        Use a unique controller listener port for each broker if required by your environment.

   3. Restart the broker with the updated configuration.

      ./bin/kafka-server-start.sh -daemon ./config/kraft/server.properties

      The broker starts with the migration-enabled configuration. Metadata migration begins after all brokers have restarted with migration enabled and have registered with the controller quorum. The time required for migration depends on the number of topics and partitions in the cluster.

   4. Verify that the broker is running.

      jcmd | grep kafka

      You should see a running kafka.Kafka process using the updated configuration file.

4. Confirm that metadata migration is complete.

   After all brokers have registered with the KRaft controller quorum, the active controller begins migrating metadata from ZooKeeper into the KRaft metadata log.

   1. Verify that the controller has entered the MIGRATION state.

      Run the following command on the active controller host:

      grep "PRE_MIGRATION to MIGRATION" ./logs/controller.log

      A log entry similar to the following indicates that the migration phase has started:

      Replayed a ZkMigrationStateRecord changing the migration state from PRE_MIGRATION to MIGRATION

   2. Check the controller logs for the completion message.

      Run the following command on the active controller host:

      grep "migration.*ZooKeeper.*KRaft" ./logs/controller.log

      A log entry similar to the following indicates that the metadata migration phase has finished:

      • Completed migration of metadata from ZooKeeper to KRaft.

   3. (Optional) Check the ZooKeeper /migration znode.

      This znode contains information about the active KRaft controller during migration:

      ./bin/zookeeper-shell.sh localhost:2181 get /migration

      The output includes the controller ID currently coordinating migration. Once the migration is complete, this information should reflect the active controller and the final metadata state.

5. Switch each broker to KRaft mode.

   After the metadata migration is complete, you must update each broker so it runs fully in KRaft mode. This step removes all ZooKeeper-specific settings and enables the broker role in KRaft.

   1. Stop the broker running on the host.

      ./bin/kafka-server-stop.sh
      jcmd | grep kafka

      If the cluster has multiple brokers, restart them one at a time.

   2. Update the broker configuration.

      Edit the server.properties file and make the following changes:

      • Replace broker.id with node.id using the same value
      • Add the KRaft broker role: process.roles=broker
      • Remove inter.broker.protocol.version
      • Remove zookeeper.metadata.migration.enable
      • Remove all ZooKeeper configuration (zookeeper.connect)

      • Remove any control-plane listener configuration (for example, control.plane.listener.name and its associated listener entry), if present.

        Example broker configuration for KRaft mode

        node.id=0
        process.roles=broker
        
        # Listener for client traffic
        listeners=PLAINTEXT://0.0.0.0:9092
        
        # Connect the broker to the controller quorum
        controller.listener.names=CONTROLLER
        controller.quorum.bootstrap.servers=controller1:9090,controller2:9090,controller3:9090

        This configuration removes all ZooKeeper dependencies and enables the standalone KRaft broker mode.

   3. Update the authorizer if you use ACLs.

      ZooKeeper-based brokers use:

      authorizer.class.name=kafka.security.authorizer.AclAuthorizer

      For KRaft-based brokers, replace it with:

      authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer

   4. Restart the broker.

      ./bin/kafka-server-start.sh -daemon ./config/kraft/server.properties

      The broker starts in KRaft mode and connects to the controller quorum.

6. Switch each controller out of migration mode.

   After all brokers are running in KRaft mode, update the controllers so that they stop interacting with ZooKeeper. This completes the migration process and places the cluster fully under KRaft control.

   1. Stop the controller.

      Stop the controller in the same way as the broker:

      ./bin/kafka-server-stop.sh
      jcmd | grep kafka

   2. Update the controller configuration.

      Edit the controller.properties file and remove all ZooKeeper-related migration settings:

      • Remove the ZooKeeper connection details
      • Remove the zookeeper.metadata.migration.enable property

      • Remove inter.broker.listener.name

        Example controller configuration after migration

        process.roles=controller
        node.id=1
        
        listeners=CONTROLLER://0.0.0.0:9090
        controller.listener.names=CONTROLLER
        listener.security.protocol.map=CONTROLLER:PLAINTEXT
        controller.quorum.bootstrap.servers=controller1:9090,controller2:9090,controller3:9090

   3. Restart the controller.

      ./bin/kafka-server-start.sh -daemon ./config/kraft/controller.properties

   4. Verify that the controller has exited migration mode.

      Check the controller log:

      grep "MIGRATION to POST_MIGRATION" ./logs/controller.log

      A log entry similar to the following indicates the transition out of migration:

      • Replayed a ZkMigrationStateRecord changing the migration state from MIGRATION to POST_MIGRATION.

        This confirms that the controller no longer relies on ZooKeeper and that the migration process is complete.

        Note

        After all brokers and controllers are running in KRaft mode and the controllers have exited migration mode, the cluster no longer relies on ZooKeeper for metadata management. If ZooKeeper is not required for other applications, you can decommission it.