Chapter 14. Scaling clusters by adding or removing nodes

Procedure

1. Configure the new broker using a broker.properties file.

   At a minimum, the broker requires the following configuration:

   • A unique node ID
   • A valid listener configuration

   • Bootstrap information for the KRaft controller quorum

     Example broker configuration

     process.roles=broker
     node.id=5
     
     listeners=PLAINTEXT://0.0.0.0:9094
     listener.security.protocol.map=PLAINTEXT:PLAINTEXT
     
     log.dirs=/var/lib/kafka/data
     controller.quorum.bootstrap.servers=node1:9093,node2:9093,node3:9093

     Copy to Clipboard Toggle word wrap

2. Format the broker:

   ./bin/kafka-storage.sh format --cluster-id <uuid> --config ./config/broker.properties --no-initial-controllers

   Copy to Clipboard Toggle word wrap

   Use the appropriate cluster ID.

3. Start the broker:

   ./bin/kafka-server-start.sh ./config/broker.properties

   Copy to Clipboard Toggle word wrap

4. Check the logs of the broker to ensure that is has successfully joined the KRaft cluster:

   tail -f ./logs/server.log

   Copy to Clipboard Toggle word wrap

5. Rebalance partitions to include the new broker using Cruise Control’s add_broker mode:

   curl -v -X POST 'http://<cruise_control_-_host>:9090/kafkacruisecontrol/add_broker?brokerid=5&dryrun=false'

   Copy to Clipboard Toggle word wrap
   Tip

   Check the rebalance with dryrun=true first to review the proposed changes before running the command with dryrun=false.

6. Monitor the rebalance progress using the Cruise Control /user_tasks endpoint:

   curl 'http://<cruise_control_host>:9090/kafkacruisecontrol/user_tasks'

   Copy to Clipboard Toggle word wrap

   Wait for the rebalance to complete.

7. Verify that the broker has been added to the active cluster:

   kafka-cluster.sh --bootstrap-server <kafka_host>:9092 --describe

   Copy to Clipboard Toggle word wrap

   This displays the list of brokers and their status. <kafka_host> must be an active broker in the cluster with a known, accessible listener port.

Procedure

1. Rebalance partitions off the broker that you want to remove using Cruise Control’s remove_broker mode:

   curl -X POST 'http://<cruise_control_-_host>:9090/kafkacruisecontrol/remove_broker?brokerid=5&dryrun=false'

   Copy to Clipboard Toggle word wrap
   Tip

   Check the rebalance with dryrun=true first to review the proposed changes before running the command with dryrun=false.

2. Monitor the rebalance progress using the Cruise Control /user_tasks endpoint:

   curl 'http://<cruise_control_host>:9090/kafkacruisecontrol/user_tasks'

   Copy to Clipboard Toggle word wrap

   Wait for the rebalance to complete.

3. Shut down the broker:

   ./bin/kafka-server-stop.sh

   Copy to Clipboard Toggle word wrap

4. Unregister the broker to remove it from the cluster metadata.

   For more information on unregistering, see Section 14.4, “Unregistering brokers after scale-down operations”.

5. Verify that the broker has been removed from the active cluster:

   kafka-cluster.sh --bootstrap-server <kafka_host>:9092 --describe

   Copy to Clipboard Toggle word wrap

   This displays the list of brokers and their status. <kafka_host> must be an active broker in the cluster with a known, accessible listener port.

1. Reassign the partitions from the broker you plan to remove to the remaining brokers using the Cruise control remove-nodes operation.
2. Update the cluster configuration, if necessary, to adjust the replication factor for topics (default.replication.factor) and the minimum required number of in-sync replica acknowledgements (min.insync.replicas).
3. Stop the Kafka broker service on the broker and remove the broker from the cluster.

Procedure

1. Unregister the removed broker from the cluster:

   ./bin/kafka-cluster.sh unregister \
     --bootstrap-server <broker_host>:<port> \
     --id <node_id_number>

   Copy to Clipboard Toggle word wrap

2. Verify the current state of the cluster by describing the topics:

   ./bin/kafka-topics.sh \
     --bootstrap-server <broker_host>:<port> \
     --describe

   Copy to Clipboard Toggle word wrap

Procedure

1. Describe the KRaft feature version.

   bin/kafka-features.sh --bootstrap-controller <controller-host>:9093 describe

   Copy to Clipboard Toggle word wrap

2. Identify the finalized KRaft version. In the output, locate the line that begins with Feature: kraft.version.

   If FinalizedVersionLevel is 0, the cluster uses a static controller quorum. If it is 1 or higher, the cluster uses a dynamic controller quorum. If the cluster uses a dynamic controller quorum, you do not need to complete any further procedures related to enabling dynamic controller quorum management.

Procedure

1. Determine whether the cluster uses a static controller quorum.

   For more information, see Checking the controller quorum type.

   If the cluster already supports dynamic controller quorum management, you do not need to complete this procedure.

2. Upgrade the KRaft feature version to enable dynamic controller quorum support.

   To upgrade all feature versions to the current release version:

   bin/kafka-features.sh --bootstrap-server <broker-host>:9092 upgrade --release-version 4.1

   Copy to Clipboard Toggle word wrap

   Alternatively, upgrade only the KRaft feature version:

   bin/kafka-features.sh --bootstrap-server <broker-host>:9092 upgrade --feature kraft.version=1

   Copy to Clipboard Toggle word wrap

3. Verify that the KRaft feature version is finalized.

   bin/kafka-features.sh --bootstrap-server <broker-host>:9092 describe

   Copy to Clipboard Toggle word wrap

   Confirm that FinalizedVersionLevel for Feature: kraft.version is 1 or higher, indicating that dynamic controller quorum management is supported.

4. Update the configuration on all brokers and controllers in the cluster.

   Remove the controller.quorum.voters property. Add the controller.quorum.bootstrap.servers property.

   For example, update the configuration as follows:

   # ... other existing properties ...
   
   # Remove the static controller quorum configuration
   # controller.quorum.voters=1@controller1:9093,2@controller2:9093,3@controller3:9093
   
   # Add the dynamic controller quorum configuration
   controller.quorum.bootstrap.servers=controller1:9093,controller2:9093,controller3:9093
   controller.listener.names=CONTROLLER

   Copy to Clipboard Toggle word wrap
5. Restart each node to apply the configuration changes.

Procedure

1. Configure a new controller node using the controller.properties file.

   At a minimum, the new controller requires the following configuration:

   • A unique node ID
   • Listener name used by the controller quorum

   • A quorum of controllers

     Example controller configuration

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

     Copy to Clipboard Toggle word wrap

     The controller.quorum.bootstrap.servers configuration includes the host and port of the new controller and each other controller already present in the cluster.

     Important

     If there is any further configuration required, such as authentication, make sure to include it in the controller.properties.

2. Update controller.quorum.bootstrap.servers in the configuration of each node in the cluster with the host and port of the new controller.

3. Set the log directory ID for the new controller:

   ./bin/kafka-storage.sh format --cluster-id <cluster_id> --config controller.properties --no-initial-controllers

   Copy to Clipboard Toggle word wrap

   By using the no-initial-controllers option, the controller is initialized without it joining the controller quorum.

4. Start the controller node

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

   Copy to Clipboard Toggle word wrap

5. Monitor the replication progress of the new controller:

   ./bin/kafka-metadata-quorum.sh --bootstrap-server localhost:9092 describe --replication

   Copy to Clipboard Toggle word wrap

   Wait until the new controller has caught up with the active controller before proceeding.

6. Add the new controller to the controller quorum:

   ./bin/kafka-metadata-quorum.sh --command-config controller.properties --bootstrap-controller localhost:9092 add-controller

   Copy to Clipboard Toggle word wrap

Procedure

1. Stop the controller node:

   ./bin/kafka-server-stop.sh

   Copy to Clipboard Toggle word wrap
2. Locate the ID of the controller and its directory ID to be able to remove it from the controller quorum. You can find this information in the meta.properties file of the metadata log.

3. Remove the controller from the controller quorum:

   ./bin/kafka-metadata-quorum.sh --bootstrap-controller localhost:9092 remove-controller --controller-id <id> --controller-directory-id <directory_id>

   Copy to Clipboard Toggle word wrap
4. Update controller.quorum.bootstrap.servers in the configuration of each node in the cluster to remove the host and port of the controller removed from the controller quorum.