Chapter 3. Getting started

1. Add a new Kafka user and group:

   groupadd kafka
   useradd -g kafka kafka
   passwd kafka

   Copy to Clipboard Toggle word wrap

   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

   Copy to Clipboard Toggle word wrap

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

   chown -R kafka:kafka /opt/kafka

   Copy to Clipboard Toggle word wrap

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

   Copy to Clipboard Toggle word wrap

   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

   Copy to Clipboard Toggle word wrap

   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

      Copy to Clipboard Toggle word wrap

      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

   Copy to Clipboard Toggle word wrap

   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

   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

   Copy to Clipboard Toggle word wrap

   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

   Copy to Clipboard Toggle word wrap

   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

   Copy to Clipboard Toggle word wrap

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

   Copy to Clipboard Toggle word wrap

   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

   Copy to Clipboard Toggle word wrap

8. Check that Kafka is running:

   jcmd | grep kafka

   Copy to Clipboard Toggle word wrap

   Returns:

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

   Copy to Clipboard Toggle word wrap

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

   tail -f ./logs/server.log

   Copy to Clipboard Toggle word wrap

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>

   Copy to Clipboard Toggle word wrap

   For example:

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

   Copy to Clipboard Toggle word wrap

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

   Copy to Clipboard Toggle word wrap

   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)

   Copy to Clipboard Toggle word wrap

   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

   Copy to Clipboard Toggle word wrap

   For example:

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

   Copy to Clipboard Toggle word wrap

   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

   Copy to Clipboard Toggle word wrap

2. Confirm that the Kafka node is stopped.

   jcmd | grep kafka

   Copy to Clipboard Toggle word wrap

1. Stop the Kafka broker:

   ./bin/kafka-server-stop.sh

   Copy to Clipboard Toggle word wrap

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/server.properties

   Copy to Clipboard Toggle word wrap

4. Check that Kafka is running:

   jcmd | grep kafka

   Copy to Clipboard Toggle word wrap

   Returns:

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

   Copy to Clipboard Toggle word wrap

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

   tail -f ./logs/server.log

   Copy to Clipboard Toggle word wrap

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

     Copy to Clipboard Toggle word wrap

     For example:

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

     Copy to Clipboard Toggle word wrap

     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
     # …

     Copy to Clipboard Toggle word wrap

     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

     Copy to Clipboard Toggle word wrap

     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.