Chapter 3. Getting started

1. Add a new Kafka user and group:

   groupadd kafka
   useradd -g kafka kafka
   passwd kafka

   This step creates the necessary user and group for managing Kafka.

2. Extract and move the contents of the amq-streams-<version>-kafka-bin.zip file and place it in the /opt/kafka directory:

   unzip amq-streams-<version>-kafka-bin.zip -d /opt
   mv /opt/kafka*redhat* /opt/kafka

3. Change the ownership of the /opt/kafka directory to the Kafka user:

   chown -R kafka:kafka /opt/kafka

4. Create the /var/lib/kafka directory for storing Kafka data and change its ownership to the Kafka user:

   mkdir /var/lib/kafka
   chown -R kafka:kafka /var/lib/kafka

   You can now run a default configuration of Kafka as a single-node cluster.

5. (Optional) Run other Kafka components, like Kafka Connect, on the same host.

   To run other components, specify the hostname and port to connect to the Kafka broker using the bootstrap.servers property in the component configuration.

   Example bootstrap servers configuration pointing to a single Kafka broker on the same host

   bootstrap.servers=localhost:9092

   However, we recommend installing and running Kafka components on separate hosts.

6. (Optional) Install Kafka components on separate hosts.

   1. Extract the Kafka installation files into the /opt/kafka/ directory on each host.
   2. Change the ownership of the /opt/kafka/ directory to the Kafka user on each host.

   3. Update the bootstrap.servers property to connect the components to the Kafka brokers running on different hosts.

      Example bootstrap servers configuration pointing to Kafka brokers on different hosts

      bootstrap.servers=kafka0.<host_ip_address>:9092,kafka1.<host_ip_address>:9092,kafka2.<host_ip_address>:9092

      You can use this configuration for Kafka Connect, MirrorMaker 2, and the Kafka Bridge.

Procedure

1. Generate a unique Kafka cluster ID using the kafka-storage tool:

   ./bin/kafka-storage.sh random-uuid

   Save this ID as it is reused for all nodes.

2. Create a configuration file for each node.

   Base the configuration file on Kafka’s provided examples:

   • controller.properties for controller-only nodes
   • broker.properties for broker-only nodes
   • server.properties for nodes with both roles
   Note

   The example server.properties file uses controller.quorum.voters by default, which is intended for static quorum setup. To use dynamic quorum (recommended), replace this with controller.quorum.bootstrap.servers, as shown in this procedure. The controller.properties file uses controller.quorum.bootstrap.servers by default.

   Adjust the following properties for each node:

   node.id=1
   process.roles=broker,controller
   controller.quorum.bootstrap.servers=node1:9093,node2:9093,node3:9093
   listeners=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
   advertised.listeners=PLAINTEXT://node1:9092
   inter.broker.listener.name=PLAINTEXT
   controller.listener.names=CONTROLLER
   log.dirs=/var/lib/kafka/data

   Important considerations:

   • Each node must have a unique node.id.
   • listeners hostname (0.0.0.0) binds to all interfaces but can be changed to a specific hostname or IP address if needed for each node.
   • advertised.listeners must reflect the actual address that clients use to connect to the Kafka node.
   • The log.dirs path in the configuration file defines where metadata is stored. If not set, it defaults to a temporary location, which is cleared on reboot and suitable only for development.

3. Bootstrap the controller quorum (one controller only):

   ./bin/kafka-storage.sh format --cluster-id <uuid> --standalone --config ./config/controller.properties

   This initializes the metadata log and bootstraps the quorum. Use it on one node only. Replace <uuid> with the generated cluster ID.

   Note

   To bootstrap with multiple controllers, use --initial-controllers instead.

4. Start the controller:

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

5. Add additional controllers to the quorum.

   Make sure controller.quorum.bootstrap.servers is correctly set across all nodes and the controllers are started.

6. Format the remaining nodes (brokers):

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

   Use the appropriate configuration file based on the role of each node. Use the same cluster ID for all nodes. This step prepares each node’s storage without modifying the controller quorum.

7. Start each broker node:

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

8. Check that Kafka is running:

   jcmd | grep kafka

   Returns:

   process ID kafka.Kafka ./config/server.properties

   Check the logs of each node to ensure that they have successfully joined the KRaft cluster:

   tail -f ./logs/server.log

Procedure

1. Start the Kafka console producer and configure it to send messages to a new topic:

   ./bin/kafka-console-producer.sh --broker-list <bootstrap_address> --topic <topic-name>

   For example:

   ./bin/kafka-console-producer.sh --broker-list localhost:9092 --topic my-topic

2. Enter several messages into the console. Press Enter to send each individual message to your new topic:

   >message 1
   >message 2
   >message 3
   >message 4

   When Kafka creates a new topic automatically, you might receive a warning that the topic does not exist:

   WARN Error while fetching metadata with correlation id 39 :
   {4-3-16-topic1=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)

   The warning should not reappear after you send further messages.

3. In a new terminal window, start the Kafka console consumer and configure it to read messages from the beginning of your new topic.

   ./bin/kafka-console-consumer.sh --bootstrap-server <bootstrap_address> --topic <topic-name> --from-beginning

   For example:

   ./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic my-topic --from-beginning

   The incoming messages display in the consumer console.

4. Switch to the producer console and send additional messages. Check that they display in the consumer console.
5. Stop the Kafka console producer and then the consumer by pressing Ctrl+C.

Procedure

1. Stop the Kafka node.

   ./bin/kafka-server-stop.sh

2. Confirm that the Kafka node is stopped.

   jcmd | grep kafka

1. Stop the Kafka broker:

   ./bin/kafka-server-stop.sh

2. Make any changes to the broker configuration that require a restart after completion.

   For further information, see the following:

   • Configuring Kafka
   • Upgrading Kafka nodes

3. Restart the Kafka broker:

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

4. Check that Kafka is running:

   jcmd | grep kafka

   Returns:

   process ID kafka.Kafka ./config/kraft/server.properties

   Check the logs of each node to ensure that they have successfully joined the KRaft cluster:

   tail -f ./logs/server.log

5. Wait until the broker has zero under-replicated partitions. You can check from the command line or use metrics.

   • Use the kafka-topics.sh command with the --under-replicated-partitions parameter:

     ./bin/kafka-topics.sh --bootstrap-server <broker_host>:<port>  --describe --under-replicated-partitions

     For example:

     ./bin/kafka-topics.sh --bootstrap-server localhost:9092 --describe --under-replicated-partitions

     The command provides a list of topics with under-replicated partitions in a cluster.

     Topics with under-replicated partitions

     Topic: topic3 Partition: 4 Leader: 2 Replicas: 2,3 Isr: 2
     Topic: topic3 Partition: 5 Leader: 3 Replicas: 1,2 Isr: 1
     Topic: topic1 Partition: 1 Leader: 3 Replicas: 1,3 Isr: 3
     # …

     Under-replicated partitions are listed if the ISR (in-sync replica) count is less than the number of replicas. If a list is not returned, there are no under-replicated partitions.

   • Use the UnderReplicatedPartitions metric:

     kafka.server:type=ReplicaManager,name=UnderReplicatedPartitions

     The metric provides a count of partitions where replicas have not caught up. You wait until the count is zero.

     Tip

     Use the Kafka Exporter to create an alert when there are one or more under-replicated partitions for a topic.