Chapter 5. Configuring Streams for Apache Kafka

Procedure

1. Add the Environment Variables Configuration Provider JAR file to the Kafka libs directory.

2. Initialize the Environment Variables Configuration Provider in the configuration properties file of the Kafka component. For example, to initialize the provider for Kafka, add the configuration to the server.properties file.

   Configuration to enable the Environment Variables Configuration Provider

   config.providers.env.class=org.apache.kafka.common.config.provider.EnvVarConfigProvider

3. Add configuration to the properties file to load data from environment variables.

   Configuration to load data from an environment variable

   option=${env:<MY_ENV_VAR_NAME>}

   Use capitalized or upper-case environment variable naming conventions, such as MY_ENV_VAR_NAME.

4. Save the changes.

5. Restart the Kafka component.

   For information on restarting brokers in a multi-node cluster, see Section 3.8, “Performing a graceful rolling restart of Kafka brokers”.

Procedure

1. Add a quorum of three new controller-only nodes.

   Integrate the controllers into the controller quorum by updating the controller.quorum.bootstrap.servers property.

   For more information, see Section 14.2, “Adding new controllers”.

2. Using the kafka-metadata-quorum.sh tool, remove the dual-role controllers from the controller quorum.

   For more information, see Section 14.3, “Removing controllers”.

3. For each dual-role node, and one at a time:

   1. Stop the dual-role node:

      ./bin/kafka-server-stop.sh

   2. Configure the dual-role node to serve as a broker-node only by switching process.roles=broker, controller in the node configuration to process.roles=broker.

      Example broker configuration

      node.id=1
      process.roles=broker
      
      log.dirs=/var/lib/kafka
      listeners=PLAINTEXT://0.0.0.0:9092
      controller.listener.names=CONTROLLER
      listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      controller.quorum.bootstrap.servers=localhost:9090, localhost:9091, localhost:9092
      inter.broker.listener.name=PLAINTEXT
      num.partitions=1
      auto.create.topics.enable=false
      default.replication.factor=3
      min.insync.replicas=2
      #...

   3. Restart the broker node that was previously serving a dual role:

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

Procedure

1. For each controller node, and one at a time:

   1. Stop the controller node:

      ./bin/kafka-server-stop.sh

   2. Configure the controller-only node to serve as a dual-role node by adding broker-specific configuration.

      At a minimum, do the following:

      • Switch process.roles=controller to process.roles=broker, controller.
      • Add or update the broker log directory using log.dirs.
      • Add a listener for the broker to handle client requests. In this example, PLAINTEXT://:9092 is added.
      • Update mappings between listener names and security protocols using listener.security.protocol.map.
      • Configure a listener for inter-broker communication using inter.broker.listener.name.

      Example dual-role configuration

      process.roles=broker, controller
      node.id=1
      
      log.dirs=/var/lib/kafka
      metadata.log.dir=/var/lib/kafka-metadata
      
      listeners=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9090
      controller.listener.names=CONTROLLER
      listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      controller.quorum.bootstrap.servers=localhost:9090
      inter.broker.listener.name=PLAINTEXT
      
      num.partitions=1
      auto.create.topics.enable=false
      default.replication.factor=3
      min.insync.replicas=2
      # ...

   3. Restart the node that is now operating in a dual role:

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

2. Use the Cruise Control remove_broker endpoint to reassign partition replicas from broker-only nodes to the nodes that now serve as dual-role nodes.

   The reassignment can take some time depending on the number of topics and partitions in the cluster.

   For more information, see Section 15.7, “Generating optimization proposals”.

3. Unregister the broker nodes:

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

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

4. Stop the broker nodes:

   ./bin/kafka-server-stop.sh