How to setup Kafka in Linux

In this article, we will learn how to setup Kafka in Ubuntu. Let’s get started.

Environment that Kafka need to work
Setup Kafka service
Create producer and consumer to test connection between Kafka cluster and Zookeeper
Create script file
Wrapping up

Environment that Kafka need to work

Ubuntu 18.4.0 LTS.
Use Putty instead of command line of WinSCP.
kafka_2.11-2.3.1

The latest version of JDK

Download jdk

  sudo apt insall default-jdk

  # hoặc
  sudo apt install default-jre

Check JDK’s version in Ubuntu

Normally, we will find the installed path of JDK with below steps:

  # update databse of ubuntu
  sudo updatedb

  # determine whether jdk is installed
  locate openjdk

Add Path of JDK into .bashrc.

  export JAVA_HOME="usr/lib/jvm/java-11-openjdk-amd64"
  export PATH=$PATH:$JAVA_HOME/bin

Cập nhật lại thay đổi trong file .bashrc file.

  source ~/.bashrc

Setup Kafka service

Run zookeeper service

Create file zookeeper.service in Ubuntu’s systemd.

  sudo nano /etc/systemd/system/zookeeper.service

Content of zookeeper.service file.

  [Unit]
  Requires=network.target remote-fs.target
  After=network.target remote-fs.target

  [Service]
  Type=simple
  User=root
  ExecStart=/home/kafka/kafka_2.11-2.3.1/bin/zookeeper-server-start.sh /home/kafka/kafka_2.11-2.3.1/config/zookeeper.properties
  ExecStop=/home/kafka/kafka_2.11-2.3.1/bin/zookeeper-server-stop.sh
  Restart=on-abnormal

  [Install]
  WantedBy=multi-user.target

Check whether zookeeper server is running.

  # setup net-tools package
  sudo apt-get install net-tools

  # zookeeper server run default at port 2181
  netstat -tulpn | grep 2181

  # stop zookeeper service
  systemctl stop zookeeper

  # if we do not install net-tools package, we can use other way to check
  telnet ip_addr port

Run zookeeper.service file.

  # run as root user
  sudo su

  # use systemctl
  systemctl start zookeeper

  # run zookeper service automatically when starting system
  systemctl enable zookeeper

Run kafka service

Create file kafka.service in systemd of Ubuntu.

  sudo nano /etc/systemd/system/kafka.service

Content of kafka.service.

  [Unit]
  Requires=zookeeper.service
  After=zookeeper.service

  [Service]
  Type=simple
  User=root
  ExecStart=/bin/sh -c '/home/kafka/kafka_2.11-2.3.1/bin/kafka-server-start.sh /home/kafka/kafka_2.11-2.3.1/config/server.properties > /home/kafka/kafka_2.11-2.3.1/kafka-logs/kafka-service-log.txt 2>&1'
  ExecStop=/home/kafka/kafka_2.11-2.3.1/bin/kafka-server-stop.sh
  Restart=on-abnormal

  [Install]
  WantedBy=multi-user.target

Check whether kafka server is running.

  # setup net-tools package
  sudo apt-get install net-tools

  # zookeeper server  run default at port 9092
  netstat -tulpn | grep 9092

  # stop zookeeper service
  systemctl stop kafka

Run kafka.service.

  # run as root user
  sudo su

  # use systemctl
  systemctl start kafka

  # run kafka service automatically when starting system
  systemctl enable kafka

Remove services that have status is failed

 sudo systemctl reset-failed

 # or
 systemctl disable service_name
 rm /etc/systemd/system/service_name
 systemctl daemon-reload

Create producer and consumer to test connection between Kafka cluster and Zookeeper

Create test topic

 # create test topic
 bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

 # check whether the test topic is created or not
 bin/kafka-topics.sh --list --zookeeper localhost:2181

Create producer

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

Create consumer

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

Create script file

```bash
sudo nano start.sh

# fill content for start.sh file
sudo systemctl reload-or-restart zookeeper.service
sudo systemctl reload-or-restart kafka.service
```

Operations for consumer groups

List all consumer groups across all topics

 bin/kafka-consumer-groups.sh --list --bootstrap-server localhost:9092

Show information about specific consumer group

 bin/kafak-consumer-groups.sh --describe --bootstrap-server localhost:9092 --group group_name

Delete specific consumer group

 bin/kafka-consumer-groups.sh --delete --bootstrap-server localhost:9092 --group group_name

Operations for topics

List all topics

 bin/kafka-topics.sh --list --zookeeper localhost:2181

Create new topic

First way, use kafka-topics.sh

  bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test

Second way, create new topic with specific consumer group name based on kafka-console-consumer.sh

  # 1. create a new consumer.properties file, such as consumer1.properties
        
  # 2. change group.id=new_group_name in consumer1.properties file

  # 3. run below command
  bin/kafka-console-consumer.sh --new-consumer --bootstrap-server localhost:9092 --topic topic_name --from-beginning --consumer.config config/consumer1.properties --delete-consumer-offsets

Other way

  # or other way
  bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --group test-consumer --topic test --from-beginning

Delete specific topic

First, we can use kafka-topics.sh to delete topic. But it only marks specific topic with marked for deletion.

  # In server.properties, set below the property
  delete.topic.enable=true

  # Then, use kafka-topics.sh to delete
  bin/kafka-topics.sh --delete --zookeeper localhost:2181 --topic topic_name

Second, delete topic completely.

  # Use Zookeeper CLI console to check whether topic will be deleted
  get /brokers/topics/topic_name

  # delete
  rmr /brokers/topics/topic_name
  rmr /admin/delete_topics/topic_name