ITcompressor IT — compressed and to the point
← Back to articles

Apache Kafka

10/06/2023
kafkastreamingbig data

Apache Kafka is a distributed streaming platform: producers write events into topics, and consumers of groups read them independently. A topic is split into partitions — the units of parallelism and ordering (messages within a partition are ordered).

Everything below revolves around a single orders-events topic with three partitions on a local broker; the cluster is covered in a separate section.

Installation#

Kafka is a JVM application: Java 17 is required. Check kafka.apache.org/downloads for the fresh version and the archive link; substitute the version numbers in the URLs below accordingly.

# Dependencies
sudo apt update
sudo apt install openjdk-17-jdk wget

# Download and unpack (Kafka 3.9, the Scala 2.13 build)
wget https://downloads.apache.org/kafka/3.9.0/kafka_2.13-3.9.0.tgz
tar -xzf kafka_2.13-3.9.0.tgz
sudo mv kafka_2.13-3.9.0 /opt/kafka

Starting (KRaft)#

Since version 3.3 Kafka runs in KRaft mode — cluster metadata is stored in Kafka's own Raft log, ZooKeeper is not needed (removed entirely in 4.0).

cd /opt/kafka

# Generate the cluster Cluster ID
KAFKA_CLUSTER_ID="$(bin/kafka-storage.sh random-uuid)"

# Format the metadata and log directory
bin/kafka-storage.sh format -t "$KAFKA_CLUSTER_ID" -c config/kraft/server.properties

# Start the broker (-daemon — background process, log in logs/server.log)
bin/kafka-server-start.sh -daemon config/kraft/server.properties

# Liveness check
tail -f /opt/kafka/logs/server.log

Kafka 3.x still supports the classic ZooKeeper mode (bin/zookeeper-server-start.sh + config/zookeeper.properties), but new deployments should be KRaft only.

Topics#

# Create a topic: 3 partitions, one replica (a single broker)
bin/kafka-topics.sh --create --topic orders-events \
  --bootstrap-server localhost:9092 --partitions 3 --replication-factor 1

# List topics
bin/kafka-topics.sh --list --bootstrap-server localhost:9092

# Description: partitions, leaders, offsets
bin/kafka-topics.sh --describe --topic orders-events --bootstrap-server localhost:9092

--partitions — how many parallel read/write streams there are (after creation partitions can only be increased). --replication-factor 1 — the only option for a single broker; in a cluster use 2–3.

Producers and consumers#

Producer#

Messages are written line by line from stdin:

bin/kafka-console-producer.sh --topic orders-events --bootstrap-server localhost:9092
> {"order_id": 1001, "amount": 149.99}
> {"order_id": 1002, "amount": 59.50}

Consumer#

# Read from the beginning of the history (--from-beginning), otherwise only new ones
bin/kafka-console-consumer.sh --topic orders-events \
  --bootstrap-server localhost:9092 --from-beginning

Consumer groups#

Consumers of the same group split the partitions among themselves: each partition is assigned to exactly one consumer of the group — this is how Kafka scales processing without losing ordering.

# A consumer from the billing group
bin/kafka-console-consumer.sh --topic orders-events \
  --bootstrap-server localhost:9092 --group billing

# Group state: assignments, current offsets, lag
bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --describe --group billing

Kafka stores group offsets itself (in the __consumer_offsets topic) — a restarted consumer continues from where it stopped.

A cluster of several brokers#

In KRaft, nodes combine broker and controller roles and elect the "meta" leader themselves. A typical production scheme is three nodes kafka1–kafka3: the roles, nodes and controller quorum are described in the server.properties of each node:

# config/kraft/server.properties, node kafka1 (on kafka2/kafka3 — node.id 2 and 3)
process.roles=broker,controller
node.id=1
controller.quorum.voters=1@kafka1.example.com:9093,2@kafka2.example.com:9093,3@kafka3.example.com:9093
listeners=PLAINTEXT://:9092,CONTROLLER://:9093
advertised.listeners=PLAINTEXT://kafka1.example.com:9092
log.dirs=/var/kafka-logs
num.partitions=3

Every node gets its own node.id, but the same controller.quorum.voters and the same Cluster ID when formatting:

bin/kafka-storage.sh format -t "$KAFKA_CLUSTER_ID" -c config/kraft/server.properties
bin/kafka-server-start.sh -daemon config/kraft/server.properties

In a cluster create topics with --replication-factor 2–3: if one broker goes down, a partition remains readable and writable thanks to the replica on another node.

Firewall#

# Clients — the consumer/producer port
sudo ufw allow from 10.0.0.0/24 to any port 9092

# Between cluster nodes — the controller port (overlapping subnets are fine)
sudo ufw allow from 10.0.0.0/24 to any port 9093

Port 2181 (ZooKeeper) no longer takes part in a KRaft deployment.

← Back to articles