
## Pulsar vs Kafka

Tldr;

![Pulsar vs Kafka](/images/pulsar-vs-kafka-metrics.png)

This tutorial walks you through deploying both Apache Pulsar and Apache Kafka on a local Kubernetes cluster using Kind, developing Go-based producers and consumers for each, performing load testing with Locust and k6 and monitoring performance using Prometheus and Grafana.

## Prerequisites

- Docker
- Kind
- kubectl
- Helm
- Go
- k6
- Locust

## Setting Up the Local Kubernetes Cluster with Kind

```bash
cat <<EOF | kind create cluster --name event-streams --config=-
kind: Cluster
apiVersion: kind.x-k8s.io/v1alpha4
nodes:
  - role: control-plane
  - role: worker
  - role: worker
EOF
```

Verify the cluster is up and running:

```bash
kubectl cluster-info --context kind-event-streams
```

## Deploy Apache Kafka (strimzi)

```bash
helm repo add strimzi https://helm-charts.strimzi.io
helm repo update
helm install strimzi-kafka-operator strimzi/strimzi-kafka-operator --namespace kafka --create-namespace

kubectl get pods -n kafka
```

## Deploy Apache Pulsar

```bash
helm repo add apache https://pulsar.apache.org/charts
helm repo update

# deploy pulsar
kubectl create namespace pulsar
helm install pulsar apache/pulsar --namespace pulsar --set initialize=true

# verify pulsar is up and running
kubectl get pods -n pulsar
```

## Kafka Producer and Consumer

We'll use the Confluent Kafka Go client.

### Producer

```golang
package main

import (
    "github.com/confluentinc/confluent-kafka-go/kafka"
    "log"
)

func main() {
    p, err := kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": "kafka.kafka.svc.cluster.local:9092"})
    if err != nil {
        log.Fatal(err)
    }
    defer p.Close()

    for i := 0; i < 10; i++ {
        p.Produce(&kafka.Message{
            TopicPartition: kafka.TopicPartition{Topic: &"test", Partition: kafka.PartitionAny},
            Value:          []byte("Hello Kafka"),
        }, nil)
    }
}
```

### Consumer

```golang
package main

import (
    "github.com/confluentinc/confluent-kafka-go/kafka"
    "log"
)

func main() {
    c, err := kafka.NewConsumer(&kafka.ConfigMap{
        "bootstrap.servers": "kafka.kafka.svc.cluster.local:9092",
        "group.id":          "go-consumer",
        "auto.offset.reset": "earliest",
    })
    if err != nil {
        log.Fatal(err)
    }
    defer c.Close()

    c.SubscribeTopics([]string{"test"}, nil)

    for {
        msg, err := c.ReadMessage(-1)
        if err == nil {
            log.Printf("Received message: %s\n", string(msg.Value))
        } else {
            log.Printf("Consumer error: %v (%v)\n", err, msg)
        }
    }
}
```

## Pulsar Producer and Consumer

Producer:

```golang
package main

import (
    "context"
    "github.com/apache/pulsar-client-go/pulsar"
    "log"
)

func main() {
    client, err := pulsar.NewClient(pulsar.ClientOptions{
        URL: "pulsar://pulsar-broker.pulsar.svc.cluster.local:6650",
    })
    if err != nil {
        log.Fatal(err)
    }
    defer client.Close()

    producer, err := client.CreateProducer(pulsar.ProducerOptions{
        Topic: "test",
    })
    if err != nil {
        log.Fatal(err)
    }
    defer producer.Close()

    if _, err := producer.Send(context.Background(), &pulsar.ProducerMessage{
        Payload: []byte("Hello Pulsar"),
    }); err != nil {
        log.Fatal(err)
    }
}

```

Consumer:

```golang
package main

import (
    "context"
    "github.com/apache/pulsar-client-go/pulsar"
    "log"
)

func main() {
    client, err := pulsar.NewClient(pulsar.ClientOptions{
        URL: "pulsar://pulsar-broker.pulsar.svc.cluster.local:6650",
    })
    if err != nil {
        log.Fatal(err)
    }
    defer client.Close()

    consumer, err := client.Subscribe(pulsar.ConsumerOptions{
        Topic:            "test",
        SubscriptionName: "go-subscriber",
        Type:             pulsar.Shared,
    })
    if err != nil {
        log.Fatal(err)
    }
    defer consumer.Close()

    msg, err := consumer.Receive(context.Background())
    if err != nil {
        log.Fatal(err)
    }
    log.Printf("Received message: %s", string(msg.Payload()))
    consumer.Ack(msg)
}
```

## Load Testing with Locust & k6

Locust:

```bash
pip3 install locust
```

Create a locustfile.py:

```python
from locust import User, task, between
from kafka import KafkaProducer

class KafkaUser(User):
    wait_time = between(1, 5)

    def on_start(self):
        self.producer = KafkaProducer(bootstrap_servers='kafka.kafka.svc.cluster.local:9092')

    @task
    def send_message(self):
        self.producer.send('test', b'Hello Kafka')

```

locust -f locustfile.py
Access the web UI at http://localhost:8089.

k6:

```bash
brew install k6
```

kafka_test.js:

```javascript
import { check } from 'k6';
import { Kafka } from 'k6/x/kafka';

const kafka = new Kafka({
    brokers: ['kafka.kafka.svc.cluster.local:9092'],
});

export default function () {
    const producer = kafka.producer();
    producer.produce({
        topic: 'test',
        messages: [{ value: 'Hello Kafka' }],
    });
    producer.disconnect();
}
```

k6 run kafka_test.js

## Monitoring with Prometheus and Grafana

```bash
helm repo add prometheus-community https://prometheus-community.github.io/helm-charts
helm repo add grafana https://grafana.github.io/helm-charts
helm repo update

kubectl create namespace monitoring

helm install prometheus prometheus-community/prometheus --namespace monitoring
helm install grafana grafana/grafana --namespace monitoring
```

Retrieve Grafana admin password:

```bash
kubectl get secret --namespace monitoring grafana -o jsonpath="{.data.admin-password}" | base64 --decode ; echo
```

Port-forward Grafana:

```bash
kubectl port-forward svc/grafana -n monitoring 3000:80
```

Access Grafana at http://localhost:3000.

## Configure Dashboards TODO

- Kafka: Use JMX Exporter to expose Kafka metrics to Prometheus.
- Pulsar: Metrics are exposed at /metrics endpoint. Ensure Prometheus is scraping these endpoints.

Import relevant dashboards into Grafana to visualize metrics like throughput, latency and resource usage.

## Comparative Analysis TODO

After setting up both systems and running load tests:

- Throughput: Measure messages per second.
- Latency: Measure the time taken for messages to be processed.
- Resource Utilization: Monitor CPU and memory usage.

Use Grafana dashboards to compare these metrics side by side for Kafka and Pulsar under similar load conditions.

Pulsar vs Kafka in K8s: Battle of Event Streams


![Kafka on K8s screenshot](/images/example.png)

# Deploying Kafka on Kubernetes with Strimzi

In this article, we'll walk through deploying an Apache Kafka cluster on a local Kubernetes cluster using [Kind](https://kind.sigs.k8s.io/) (Kubernetes in Docker). We'll use the [Strimzi operator](https://strimzi.io/) for Kafka, which simplifies the deployment and management of Kafka on Kubernetes. We'll also touch on how you could incorporate Terraform to manage some of the setup or perform upgrades.

## Prerequisites

- **Docker Desktop** installed
- **Kind** installed
- **kubectl** installed

## 1. Create a Local Kubernetes Cluster with Kind

First, create a Kind cluster. For a simple default cluster:

```bash
kind create cluster --name kafka-demo
```

Verify your cluster:

```bash
kubectl cluster-info --context kind-kafka-demo
```


## 2. Install the Strimzi Operator

2.1 Using kubectl
Strimzi provides YAML manifests you can apply directly:

```bash
# Download the Strimzi installation YAML
curl -L "https://github.com/strimzi/strimzi-kafka-operator/releases/download/0.34.0/strimzi-cluster-operator-0.34.0.yaml" -o strimzi.yaml
```

# Apply to your cluster

```bash
kubectl create namespace kafka
kubectl apply -f strimzi.yaml -n kafka
```

## 2.2 Using Helm (Optional)

```bash
helm repo add strimzi https://strimzi.io/charts/
helm repo update
helm install strimzi-kafka-operator strimzi/strimzi-kafka-operator --namespace kafka --create-namespace
```

## 1. Deploy a Kafka Cluster

Create a kafka-cluster.yaml file:

```yaml
apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: my-kafka-cluster
  namespace: kafka
spec:
  kafka:
    replicas: 3
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
    storage:
      type: ephemeral
  zookeeper:
    replicas: 3
    storage:
      type: ephemeral
  entityOperator:
    topicOperator: {}
    userOperator: {}
```

```
kubectl apply -f kafka-cluster.yaml -n kafka
```

Strimzi will spin up your Kafka brokers and Zookeeper pods.


## 1. Verifying Kafka

Once the operator has created the Kafka cluster, check your pods:

```bash

kubectl get pods -n kafka

You should see:
```bash
                                   READY   STATUS    RESTARTS   AGE
my-kafka-cluster-kafka-0                1/1     Running   0          2m
my-kafka-cluster-kafka-1                1/1     Running   0          2m
my-kafka-cluster-kafka-2                1/1     Running   0          2m
my-kafka-cluster-zookeeper-0            1/1     Running   0          2m
my-kafka-cluster-zookeeper-1            1/1     Running   0          2m
my-kafka-cluster-zookeeper-2            1/1     Running   0          2m
strimzi-cluster-operator-...            1/1     Running   0          3m
```

## 6. Testing Kafka

You can create a Kafka topic and produce/consume messages:

```bash
# Create a topic

kubectl exec -it my-kafka-cluster-kafka-0 -n kafka -- \
  bin/kafka-topics.sh --create --topic test-topic --partitions 1 --replication-factor 1 --bootstrap-server my-kafka-cluster-kafka-bootstrap:9092

# Produce messages
kubectl exec -it my-kafka-cluster-kafka-0 -n kafka -- \
  bin/kafka-console-producer.sh --topic test-topic --bootstrap-server my-kafka-cluster-kafka-bootstrap:9092

# Consume messages
kubectl exec -it my-kafka-cluster-kafka-1 -n kafka -- \
  bin/kafka-console-consumer.sh --topic test-topic --from-beginning --bootstrap-server my-kafka-cluster-kafka-bootstrap:9092

Type a few messages in the producer terminal, and they should appear in the consumer terminal.
```

## 7. Clean Up

When you’re done, remove the cluster:

```bash
kind delete cluster --name kafka-demo
```

## Conclusion

With this setup, you have a quick local environment for developing or testing Kafka. The Strimzi operator abstracts away the complexity of Kafka management on Kubernetes. If you want to scale to a production environment, you’ll switch from ephemeral storage to persistent volumes, integrate with your cloud platform, and further tune your Terraform and Strimzi configs.


A step-by-step guide to setting up a Kafka cluster on a local Kind cluster using the Strimzi operator, with optional Terraform provisioning.

Posts tagged with: kafka