Install Kafka

What is Kafka

Apache Kafka is a distributed event store and stream-processing platform. It is an open-source system developed by the Apache Software Foundation and written in Java and Scala. The project aims to provide a unified, high-throughput, low-latency platform for handling real-time data feeds.

Install Kafka

💡 Your local environment must have Java 8+ installed.

Downloading and extracting the Kafka Binaries

Download the latest Kafka release and extract it:

curl "<https://dlcdn.apache.org/kafka/3.4.0/kafka_2.13-3.4.0.tgz>" -o ~/Downloads/kafka.tgz
# Or download this version
# curl "<https://archive.apache.org/dist/kafka/2.8.2/kafka_2.12-2.8.2.tgz>" -o ~/Downloads/kafka2.tgz

mkdir ~/kafka && cd ~/kafka
#You specify the --strip 1 flag to ensure that the archive's contents are extracted in ~/kafka/ itself and not in another directory
tar -xvzf ~/Downloads/kafka.tgz --strip 1

I’m installing Kafka in Ubuntu 22

Clean Previous Configuration

Stop Service:

service zookeeper stop
service kafka stop

Clean kafka and zookeeper data:

# Some files may not exist
sudo rm -rf /tmp/zookeeper
sudo rm -rf /tmp/kafka-logs
sudo rm -rf /etc/systemd/system/zookeeper.service
sudo rm -rf /etc/systemd/system/multi-user.target.wants/zookeeper.service
sudo rm -rf /etc/systemd/system/multi-user.target.wants/kafka.service
sudo rm -rf /etc/systemd/system/kafka.service

systemctl daemon-reload

Start Kafka without Auth

if you want to use Kafka with Auth skip this section and follow Configuring the Kafka Server with Auth section.

cd /home/kafka

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

Follow the Manage Topic section to test Kafka.

Configuring the Kafka Server with Auth

Config server.properties

Kafka’s configuration options are specified in server.properties. Open this file with the nano or your favorite editor:

nano /home/kafka/config/server.properties

First, add a setting that will allow you to delete Kafka topics. Add the following line to the bottom of the file:

delete.topic.enable = true

and add this at the end for authorization to enable authentication:

security.inter.broker.protocol=SASL_PLAINTEXT
sasl.mechanism.inter.broker.protocol=PLAIN
sasl.enabled.mechanisms=PLAIN

authorizer.class.name=kafka.security.auth.SimpleAclAuthorizer
allow.everyone.if.no.acl.found=true
listeners=SASL_PLAINTEXT://0.0.0.0:9092
advertised.listeners=SASL_PLAINTEXT://:9092

💡 If you can not connect to Kafka remotely you should make this change:

# This is important to connect from outside
advertised.listeners=SASL_PLAINTEXT://YOUR_EXTERNAL_IP_HEAR:9092

Save and close the file.

Config zookeeper_jaas.conf

Create zookeeper_jaas.conf file and set config:

cd /home/kafka 
nano config/zookeeper_jaas.conf

And add this:

Server {
    org.apache.kafka.common.security.plain.PlainLoginModule required
    username="admin"
    password="123456"
    user_admin="123456";
};

Config kafka_server_jaas.conf

Create kafka_server_jaas.conf file and set config:

cd /home/kafka 
nano config/kafka_server_jaas.conf

And add this:

KafkaServer {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  username="admin"
  password="123456"
  user_admin="123456";
};

Client {
  org.apache.kafka.common.security.plain.PlainLoginModule required
  username="admin"
  password="123456";
};

Config zookeeper.properties

run nano config/zookeeper.properties

authProvider.1=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasl
jaasLoginRenew=3600000

KAFKA_OPTS in zookeeper-server-start.sh

you have to add KAFKA_OPTS to .sh or systemd service:

add the bellow code to “zookeeper-server-start.sh”

# Note: Change file address
export KAFKA_OPTS="-Djava.security.auth.login.config=/home/kafka/config/zookeeper_jaas.conf"

or add the bellow code to /etc/systemd/system/zookeeper.service:

Environment="KAFKA_OPTS=-Djava.security.auth.login.config=/home/kafka/config/zookeeper_jaas.conf"

KAFKA_OPTS in kafka-server-start.sh

you have to add KAFKA_OPTS to .sh or systemd service:

add the bellow code to “kafka-server-start.sh”

# Note: Change file address 
export KAFKA_OPTS="-Djava.security.auth.login.config=/home/kafka/config/kafka_server_jaas.conf"

or add the bellow code to /etc/systemd/system/zookeeper.service:

Environment="KAFKA_OPTS=-Djava.security.auth.login.config=/home/kafka/config/kafka_server_jaas.conf"

Start Kafka

Apache Kafka could start with a bash file or with a service

Create Kafka Service

We want to create systemed Unit Files and Start the Kafka Server and Zookeeper Server.

These files will help you perform common service actions such as starting, stopping, and restarting Kafka in a manner consistent with other Linux services.

Kafka uses Zookeeper to manage its cluster state and configurations. You’ll use Zookeeper as a service with these unit files.

Create the unit file for the zookeeper:

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

Enter the following unit definition into the file:

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

[Service]
Type=simple
User=root
Environment="KAFKA_OPTS=-Djava.security.auth.login.config=/home/kafka/config/zookeeper_jaas.conf"
ExecStart=/home/kafka/bin/zookeeper-server-start.sh /home/kafka/config/zookeeper.properties
ExecStop=/home/kafka/bin/zookeeper-server-stop.sh
Restart=on-abnormal

[Install]
WantedBy=multi-user.target

• The [Unit] section specifies that Zookeeper requires networking and the filesystem to be ready before starting.
• The [Service] section specifies that systemd should use the zookeeper-server-start.sh and zookeeper-server-stop.sh shell files for starting and stopping the service. It also specifies that Zookeeper should be restarted if it exits abnormally.

After adding this content, save and close the file.

Next, create the systemd service file for kafka:

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

[Service]
Type=simple
User=root
Environment="KAFKA_OPTS=-Djava.security.auth.login.config=/home/kafka/config/kafka_server_jaas.conf"
ExecStart=/bin/sh -c '/home/kafka/bin/kafka-server-start.sh /home/kafka/config/server.properties > /home/kafka/kafka.log 2>&1'
ExecStop=/home/kafka/bin/kafka-server-stop.sh
Restart=on-abnormal

[Install]
WantedBy=multi-user.target

The following unit definition into the file:

The [Unit] section specifies that this unit file depends on zookeeper.service, which will ensure that zookeeper gets started automatically when the kafka service starts.

The [Service] section specifies that systemd should use the kafka-server-start.sh and kafka-server-stop.sh shell files for starting and stopping the service. It also specifies that Kafka should be restarted if it exits abnormally.

Now that you have defined the units, start Kafka with the following command:

systemctl daemon-reload

# Start Service
sudo systemctl start kafka.service

To ensure that the server has started successfully, check the journal logs for the kafka unit:

sudo systemctl status kafka.service

You will receive output like this:

Kafka status

You now have a Kafka server listening on port 9092, which is the default port the Kafka server uses.

You have started the kafka service. But if you reboot your server, Kafka will not restart automatically. To enable the kafka service on server boot, run the following command:

sudo systemctl start zookeeper.service

Then run this command:

sudo systemctl enable kafka.service
sudo  systemctl enable zookeeper.service

In this step, you started and enabled the kafka and zookeeper services. In the next step, you will check the Kafka installation.

Manage Topics

Create Topic

Before you can write your first events, you must create a topic. Open another terminal session and run:

# create Topic "test"
bin/kafka-topics.sh --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic test

All of Kafka’s command line tools have additional options: run the kafka-topics.sh command without any arguments to display usage information. For example, it can also show you details such as the partition count of the new topic:

bin/kafka-topics.sh --describe --topic test --bootstrap-server localhost:9092

# Show Results like this:
# Topic: quickstart-events        TopicId: NPmZHyhbR9y00wMglMH2sg PartitionCount: 1       ReplicationFactor: 1	Configs:
#    Topic: quickstart-events Partition: 0    Leader: 0   Replicas: 0 Isr: 0

Write on Topic

A Kafka client communicates with the Kafka brokers via the network for writing (or reading) events. Once received, the brokers will store the events in a durable and fault-tolerant manner for as long as you need — even forever.

Run the console producer client to write a few events into your topic. By default, each line you enter will result in a separate event being written to the topic.

# post to the topic "test"
bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

# This is my first event
# This is my second event

You can stop the producer client with Ctrl-C at any time.

Read the Event

Open another terminal session and run the console consumer client to read the events you just created:

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

# This is my first event
# This is my second event

Test With Sample Code

You can Test your kafka server with kafkajs client. There is an sample source code in the github.

References

• How To Install Apache Kafka on Ubuntu 20.04 | DigitalOcean
• Apache Kafka
• Docker
• How to Set Up Authentication in Kafka Cluster | CodeForGeek
• IBM Developer