Chapter 2. Getting started

Procedure

1. Add new kafka user and group.

   sudo groupadd kafka
   sudo useradd -g kafka kafka
   sudo passwd kafka

2. Create directory /opt/kafka.

   sudo mkdir /opt/kafka

3. Create a temporary directory and extract the contents of the AMQ Streams ZIP file.

   mkdir /tmp/kafka
   unzip amq-streams_y.y-x.x.x.zip -d /tmp/kafka

4. Move the extracted contents into /opt/kafka directory and delete the temporary directory.

   sudo mv /tmp/kafka/kafka_y.y-x.x.x/* /opt/kafka/
   rm -r /tmp/kafka

5. Change the ownership of the /opt/kafka directory to the kafka user.

   sudo chown -R kafka:kafka /opt/kafka

6. Create directory /var/lib/zookeeper for storing ZooKeeper data and set its ownership to the kafka user.

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

7. Create directory /var/lib/kafka for storing Kafka data and set its ownership to the kafka user.

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

Running the cluster

1. Edit the ZooKeeper configuration file /opt/kafka/config/zookeeper.properties. Set the dataDir option to /var/lib/zookeeper/:

   dataDir=/var/lib/zookeeper/

2. Edit the Kafka configuration file /opt/kafka/config/server.properties. Set the log.dirs option to /var/lib/kafka/:

   log.dirs=/var/lib/kafka/

3. Switch to the kafka user:

   su - kafka

4. Start ZooKeeper:

   /opt/kafka/bin/zookeeper-server-start.sh -daemon /opt/kafka/config/zookeeper.properties

5. Check that ZooKeeper is running:

   jcmd | grep zookeeper

   Returns:

   number org.apache.zookeeper.server.quorum.QuorumPeerMain /opt/kafka/config/zookeeper.properties

6. Start Kafka:

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

7. Check that Kafka is running:

   jcmd | grep kafka

   Returns:

   number kafka.Kafka /opt/kafka/config/server.properties

Procedure

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

   /opt/kafka/bin/kafka-console-producer.sh --broker-list <bootstrap-address> --topic <topic-name>

   For example:

   /opt/kafka/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.

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

   For example:

   /opt/kafka/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 broker.

   su - kafka
   /opt/kafka/bin/kafka-server-stop.sh

2. Confirm that the Kafka broker is stopped.

   jcmd | grep kafka

3. Stop ZooKeeper.

   su - kafka
   /opt/kafka/bin/zookeeper-server-stop.sh

Procedure

1. Open ZooKeeper and Kafka broker configuration files in a text editor. The configuration files are located at :

   ZooKeeper

   /opt/kafka/config/zookeeper.properties

   Kafka

   /opt/kafka/config/server.properties

2. Edit the configuration options. The configuration files are in the Java properties format. Every configuration option should be on separate line in the following format:

   <option> = <value>

   Lines starting with # or ! will be treated as comments and will be ignored by AMQ Streams components.

   # This is a comment

   Values can be split into multiple lines by using \ directly before the newline / carriage return.

   sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
       username="bob" \
       password="bobs-password";

3. Save the changes
4. Restart the ZooKeeper or Kafka broker
5. Repeat this procedure on all the nodes of the cluster.