# Kafka Basics

> [Apache Kafka](https://kafka.apache.org/) is an open-source distributed event streaming platform for high-performance data pipelines, streaming analytics, data integration, and mission-critical applications.

## Introduction

This article aims to provide basic hands-on instructions on how to interact with Kafka in console.

## Installation

1. [Download the binary distribution of Apache Kafka](https://kafka.apache.org/downloads).
    
2. Extract Kafka (this document is based on Kafka version 2.13-3.6.1).
    
    ```shell
    tar -xzf kafka_2.13-3.6.1.tgz
    cd kafka_2.13-3.6.1
    ```
    

## Start Kafka

There are two ways to start a Kafka.

### Start Kafka with ZooKeeper

To start a Kafka cluster using **ZooKeeper**, open a Terminal and follow the commands below.

```shell
export KAFKA_HOME=<Directory of your Kafka distribution>
$KAFKA_HOME/bin/zookeeper-server-start.sh config/zookeeper.properties
```

### Start Kafka with Kraft

To start a Kafka cluster using **KRaft**, open a Terminal and follow the commands below.

1. Format the storage directory:
    
    ```shell
    export KAFKA_HOME=<Directory of your Kafka distribution>
    export KAFKA_CLUSTER_ID=`$KAFKA_HOME/bin/kafka-storage.sh random-uuid`
    $KAFKA_HOME/bin/kafka-storage.sh format -t $KAFKA_CLUSTER_ID -c $KAFKA_HOME/config/kraft/server.properties
    ```
    
2. Start Kafka using KRaft.
    
    ```shell
    $KAFKA_HOME/bin/kafka-server-start.sh $KAFKA_HOME/config/kraft/server.properties
    ```
    
3. To terminate, input `CTRL+C`.
    

## Kafka CLI

This section shows the common Kafka CLI commands for Kafra operations.

### Get the cluster metadata

1. Open another Terminal. Set environment variables.
    
    ```shell
    export KAFKA_HOME=<Directory of your Kafka distribution>
    export BOOTSTRAP_SERVER=localhost:9092
    ```
    
2. List cluster ID. kafka-metadata-quorum.sh is used to describe the runtime state of the cluster metadata partition, including cluster ID.
    
    ```shell
    $KAFKA_HOME/bin/kafka-metadata-quorum.sh --bootstrap-server $BOOTSTRAP_SERVER describe --status
    ```
    
    Here below lists an example of the output:
    
    ```text
    ClusterId:              STyWMJVBQQyaSt7WEgs5nQ
    LeaderId:               1
    LeaderEpoch:            1
    HighWatermark:          739
    MaxFollowerLag:         0
    MaxFollowerLagTimeMs:   0
    CurrentVoters:          [1]
    CurrentObservers:       []
    ```
    

### List all topics

```shell
$KAFKA_HOME/bin/kafka-topics.sh --bootstrap-server $BOOTSTRAP_SERVER --list
```

### Create a Topic

```shell
export TOPIC_NAME=test-topic
$KAFKA_HOME/bin/kafka-topics.sh --create --bootstrap-server $BOOTSTRAP_SERVER --replication-factor 1 --partitions 1 --topic $TOPIC_NAME
```

### Start Kafka Console Producer

The following command starts a Kafka producer, using comma as the separator between the key and value. Note that when a topic

```shell
$KAFKA_HOME/bin/kafka-console-producer.sh --broker-list $BOOTSTRAP_SERVER \
    --topic $TOPIC_NAME \
    --property parse.key=true \
    --property key.separator=,
```

A `>` prompt should be shown in the next line. Input a pair of key and value text separated by a comma, e.g. `Hello,World`.

![](https://cdn.hashnode.com/res/hashnode/image/upload/v1705594713739/7214d14a-5134-4701-8862-b912ec4d4f90.png align="left")

To terminate, input `CTRL+C`.

### Start Kafka Console Consumer

1. Open another Terminal. Set environment variables.
    
    ```shell
    export KAFKA_HOME=<Directory of your Kafka distribution>
    export BOOTSTRAP_SERVER=localhost:9092
    export TOPIC_NAME=test-topic
    ```
    
2. Start a Kafka Console Consumer.
    
    ```shell
    $KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server $BOOTSTRAP_SERVER \
        --topic $TOPIC_NAME \
        --from-beginning \
        --property print.key=true \
        --property print.value=true
    ```
    
3. The console should show the messages in the topic from the beginning.
    
    ![](https://cdn.hashnode.com/res/hashnode/image/upload/v1705594333054/e3186256-4c9d-4f46-94e3-a48016bf5963.png align="left")
    
4. To terminate, input `CTRL+C`.
    

## Kafka Messages with header

Apache Kafka 0.11 introduced the concept of record headers.

### Start Kafka Console Producer

By default, the header delimiter, which separates the headers and key/value pair, is "\\t" and the key separator, which separate the key and the value, is also "\\t".

The following Kafka Producer Console commands start the producer with the header delimiter and the key separator changed to semicolon and comma respectively.

```shell
export KAFKA_HOME=<Directory of your Kafka distribution>
export BOOTSTRAP_SERVER=localhost:9092
export TOPIC_NAME=test-headers-topic

$KAFKA_HOME/bin/kafka-console-producer.sh --bootstrap-server $BOOTSTRAP_SERVER \
--topic $TOPIC_NAME \
--property parse.headers=true \
--property parse.key=true \
--property key.separator=, \
--property headers.delimiter=\;
```

### Start Kafka Console Consumer

Open another Terminal. Start a Kafka Console Consumer, which consumes the using following commands.

```shell
export KAFKA_HOME=<Directory of your Kafka distribution>
export BOOTSTRAP_SERVER=localhost:9092
export TOPIC_NAME=test-headers-topic

$KAFKA_HOME/bin/kafka-console-consumer.sh --bootstrap-server $BOOTSTRAP_SERVER \
--topic $TOPIC_NAME \
--from-beginning \
--property print.headers=true \
--property print.key=true \
--property print.value=true
```

### Publish test messages with headers

Send the following test message from the Kafka Console Producer.

```shell
h0-k1:h0-v1,h0-k2:h0-v2;key 0,value 0
h1-k1:h1-v1,h1-k2:h1-v2;key 1,value 1
h2-k1:h2-v1,h2-k2:h2-v2;key 2,value 2
```

![](https://cdn.hashnode.com/res/hashnode/image/upload/v1705594777591/2ad66cdf-8b2c-4777-ac31-b6239da8312d.png align="left")

### Receive test messages with headers

Test messages with headers should be received from the Kafka Console Consumer.

![](https://cdn.hashnode.com/res/hashnode/image/upload/v1705594792244/c378d996-ac00-4790-82b8-83f3461da1ed.png align="left")