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. Start ZooKeeper.

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

3. Make sure that Apache ZooKeeper is running.

   jcmd | grep zookeeper

   Returns:

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

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

   log.dirs=/var/lib/kafka/

5. Start Kafka.

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

6. Make sure that Kafka is running.

   jcmd | grep kafka

   Returns:

   number kafka.Kafka config/server.properties

Procedure

1. Start the Kafka console producer.

   /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. Type your message into the console where the producer is running.

3. Press Enter to send.

   When Kafka creates a new topic automatically, you might receive warning messages to say 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)
   
   WARN [Producer clientId=console-producer] Error while fetching metadata with correlation id 4 :
   {my-topic=LEADER_NOT_AVAILABLE} (org.apache.kafka.clients.NetworkClient)

   The warning messages should not reappear after you start sending messages to the new topic.

   The auto.create.topics.enable configuration property enables the automatic creation of topics.

4. Start the message receiver.

   /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

5. Confirm that you see the incoming messages in the consumer console.
6. Press Crtl+C to exit the Kafka console producer and consumer.

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.