Kafka #

Apache Kafka is a distributed event streaming platform designed to handle millions of messages per second with low latency and high durability. Unlike traditional message brokers like RabbitMQ that focus on message routing, Kafka is designed as a persistent distributed log — messages are stored on disk and can be re-read independently by many consumers. This makes it ideal for event sourcing, real-time data pipelines, audit logs, and large-scale microservice integration. Understanding the concepts of Topic, Partition, Consumer Group, and Offset is the foundation before writing reliable Kafka code.

Kafka Basic Concepts #

flowchart LR
    subgraph Producers["Producer"]
        A["App A"]
        B["App B"]
    end

    subgraph Broker["Broker (Kafka Server)"]
        subgraph Topic["Topic: orders"]
            P0["Partition 0: [msg1][msg2][msg3]"]
            P1["Partition 1: [msg4][msg5]"]
            P2["Partition 2: [msg6][msg7][msg8]"]
        end
        Note["A topic can have many partitions\nfor parallelism and scalability"]
        Topic -.- Note
    end

    subgraph Consumers["Consumer"]
        subgraph GroupA["Group A"]
            X["Service X"]
            Y["Service Y"]
        end
    end

    A --> |publish| Topic
    B --> |publish| Topic
    Topic --> X
    Topic --> Y

Important concepts:

Topic: message category/channel (like a “table” in a database)
Partition: a topic subdivision for parallelism; messages within one partition are ordered
Offset: a message’s position within a partition (0, 1, 2, …)
Consumer Group: a group of consumers sharing the load of reading one topic
Broker: a Kafka server; a cluster usually consists of 3+ brokers
Replication: every partition has N replicas on different brokers for fault tolerance

Since Kafka 3.3+, KRaft mode (Kafka Raft) is available as a Zookeeper replacement and is the default in recent versions. KRaft significantly simplifies cluster operations — you no longer need to run Zookeeper separately. For new installations, use Kafka 3.x with KRaft mode.

Installation #

# Python library for Kafka
pip install kafka-python

# Alternative: confluent-kafka (librdkafka binding, higher performance for production)
pip install confluent-kafka

To run Kafka locally, the easiest way is Docker:

# docker-compose.yml for Kafka with KRaft (no Zookeeper)
# Save as docker-compose.yml and run: docker compose up -d

version: "3"
services:
  kafka:
    image: bitnami/kafka:latest
    ports:
      - "9092:9092"
    environment:
      - KAFKA_CFG_NODE_ID=1
      - KAFKA_CFG_PROCESS_ROLES=broker,controller
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092
      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      - KAFKA_CFG_CONTROLLER_QUORUM_VOTERS=1@kafka:9093
      - KAFKA_CFG_CONTROLLER_LISTENER_NAMES=CONTROLLER

Producer — Sending Messages #

A producer sends messages to a Kafka topic. A message consists of a key (optional) and a value, both as bytes. JSON serialization must be done explicitly.

import json
import os
from kafka import KafkaProducer
from kafka.errors import KafkaError

BOOTSTRAP_SERVERS = os.getenv("KAFKA_BOOTSTRAP_SERVERS", "localhost:9092")

# ANTI-PATTERN: sending raw bytes without serialization
producer = KafkaProducer(bootstrap_servers=BOOTSTRAP_SERVERS)
producer.send("orders", b"{'order_id': 1}")  # ✗ -- not valid JSON, hard to consume

# CORRECT: configure a JSON serializer
producer = KafkaProducer(
    bootstrap_servers=BOOTSTRAP_SERVERS,
    value_serializer=lambda v: json.dumps(v, ensure_ascii=False).encode("utf-8"),
    key_serializer=lambda k: k.encode("utf-8") if k else None,
    acks="all",             # wait for confirmation from all replicas (safest)
    retries=3,              # retry on failure
    retry_backoff_ms=300,   # delay between retries
    request_timeout_ms=30000,
    compression_type="gzip" # compress messages for bandwidth efficiency
)

Sending a Simple Message #

def kirim_order(producer: KafkaProducer, order: dict) -> None:
    topic   = "orders"
    # The key determines the partition -- messages with the same key always go
    # to the same partition, preserving order per entity
    # Use the entity ID as the key to keep ordering per entity
    key     = str(order.get("order_id"))

    future = producer.send(topic, key=key, value=order)

    try:
        record_metadata = future.get(timeout=10)  # wait for confirmation
        print(
            f"Message sent → topic: {record_metadata.topic}, "
            f"partition: {record_metadata.partition}, "
            f"offset: {record_metadata.offset}"
        )
    except KafkaError as e:
        print(f"Failed to send message: {e}")

# Usage
order = {
    "order_id":   1001,
    "pengguna_id": 42,
    "produk":     "Laptop Gaming ASUS",
    "total":      18500000,
    "status":     "pending",
    "timestamp":  "2024-03-15T10:30:00Z"
}

kirim_order(producer, order)
producer.flush()  # make sure all messages are sent before exiting

Producer with Callbacks (Non-blocking) #

def on_send_success(record_metadata):
    print(
        f"✓ Sent → {record_metadata.topic}:"
        f"[{record_metadata.partition}]@{record_metadata.offset}"
    )

def on_send_error(exc):
    print(f"✗ Failed to send: {exc}")

def kirim_event_async(producer: KafkaProducer, topic: str, key: str, event: dict) -> None:
    producer.send(topic, key=key, value=event) \
            .add_callback(on_send_success)     \
            .add_errback(on_send_error)

# Send many events asynchronously
events = [
    {"event": "user_login",    "user_id": 1, "ip": "192.168.1.1"},
    {"event": "product_view",  "user_id": 1, "product_id": 101},
    {"event": "add_to_cart",   "user_id": 1, "product_id": 101},
]

for event in events:
    kirim_event_async(producer, "user-events", str(event["user_id"]), event)

producer.flush()  # wait for all callbacks to finish
producer.close()

Consumer — Receiving Messages #

A consumer reads messages from a topic. The auto_offset_reset and enable_auto_commit settings are the two parameters that most often become sources of bugs.

from kafka import KafkaConsumer
from kafka.errors import KafkaError
import json
import signal

BOOTSTRAP_SERVERS = os.getenv("KAFKA_BOOTSTRAP_SERVERS", "localhost:9092")

# ANTI-PATTERN: enable_auto_commit=True without understanding the risks
consumer = KafkaConsumer(
    "orders",
    bootstrap_servers=BOOTSTRAP_SERVERS,
    enable_auto_commit=True,   # ✗ -- auto-commits before processing finishes
    group_id="order-service"   #    can cause message loss if it crashes mid-process
)

# CORRECT: manual commit after successful processing
consumer = KafkaConsumer(
    "orders",
    bootstrap_servers=BOOTSTRAP_SERVERS,
    group_id="order-service",
    auto_offset_reset="earliest",   # start from the beginning if the group is new / no offset
                                    # "latest" = only read new messages
    enable_auto_commit=False,       # manual commit after successful processing
    value_deserializer=lambda v: json.loads(v.decode("utf-8")),
    key_deserializer=lambda k: k.decode("utf-8") if k else None,
    session_timeout_ms=30000,
    heartbeat_interval_ms=10000,
    max_poll_records=50             # maximum messages per poll
)

Consumer Loop with Graceful Shutdown #

import signal
import threading

def proses_order(order: dict) -> None:
    """Order processing business logic."""
    print(f"Processing order #{order['order_id']} — total Rp{order['total']:,.0f}")
    # ... save to the database, send notifications, etc.

def jalankan_consumer():
    berjalan = True

    def handle_shutdown(signum, frame):
        nonlocal berjalan
        print("Received shutdown signal, stopping consumer...")
        berjalan = False

    signal.signal(signal.SIGINT,  handle_shutdown)
    signal.signal(signal.SIGTERM, handle_shutdown)

    consumer = KafkaConsumer(
        "orders",
        bootstrap_servers=BOOTSTRAP_SERVERS,
        group_id="order-service",
        auto_offset_reset="earliest",
        enable_auto_commit=False,
        value_deserializer=lambda v: json.loads(v.decode("utf-8")),
        key_deserializer=lambda k: k.decode("utf-8") if k else None,
    )

    print("Consumer active, waiting for messages...")
    try:
        while berjalan:
            # poll() with a timeout so the running flag can be checked periodically
            records = consumer.poll(timeout_ms=1000, max_records=50)

            for topic_partition, messages in records.items():
                for msg in messages:
                    try:
                        proses_order(msg.value)
                        # Commit only after successful processing
                        consumer.commit()
                    except Exception as e:
                        print(f"Error processing message at offset {msg.offset}: {e}")
                        # Don't commit -- the message will be re-read after a restart
                        # Add it to a dead letter queue if needed

    finally:
        consumer.close()
        print("Consumer shut down cleanly.")

jalankan_consumer()

Consumer Groups and Partition Assignment #

A consumer group lets several consumer instances share the load of reading one topic — each partition is read by only one consumer in the group at a time.

# Consumer group diagram:
#
# Topic "orders" with 3 partitions:
#
# Partition 0 ──────► Consumer A  ┐
# Partition 1 ──────► Consumer B  ├── Group: "order-service"
# Partition 2 ──────► Consumer C  ┘
#
# If Consumer B dies:
# Partition 0 ──────► Consumer A  ┐
# Partition 1 ──────► Consumer A  ├── Automatic rebalance
# Partition 2 ──────► Consumer C  ┘

# Subscribe to several topics at once
consumer.subscribe(["orders", "payments", "notifications"])

# Or manually assign specific partitions (no rebalance)
from kafka import TopicPartition

consumer.assign([
    TopicPartition("orders", 0),
    TopicPartition("orders", 1),
])

# Check the current assignment
print("Assigned partitions:", consumer.assignment())

# Check the lag (how many messages are unprocessed)
for tp in consumer.assignment():
    committed = consumer.committed(tp)
    end       = consumer.end_offsets([tp])[tp]
    lag       = end - (committed.offset if committed else 0)
    print(f"{tp.topic}[{tp.partition}]: lag={lag}")

Message Serialization #

For production applications, use formats more efficient than JSON — such as Avro or Protobuf — especially for high throughput.

import json
from dataclasses import dataclass, asdict
from datetime import datetime

@dataclass
class OrderEvent:
    order_id:   int
    user_id:    int
    total:      float
    status:     str
    created_at: str = None

    def __post_init__(self):
        if not self.created_at:
            self.created_at = datetime.utcnow().isoformat() + "Z"

# Producer with a dataclass
producer = KafkaProducer(
    bootstrap_servers=BOOTSTRAP_SERVERS,
    value_serializer=lambda v: json.dumps(asdict(v), ensure_ascii=False).encode("utf-8"),
    key_serializer=lambda k: str(k).encode("utf-8") if k else None,
    acks="all"
)

event = OrderEvent(order_id=1001, user_id=42, total=18500000.0, status="created")
producer.send("order-events", key=event.order_id, value=event)
producer.flush()

# Consumer that reconstructs the dataclass
consumer = KafkaConsumer(
    "order-events",
    bootstrap_servers=BOOTSTRAP_SERVERS,
    group_id="order-processor",
    auto_offset_reset="earliest",
    enable_auto_commit=False,
    value_deserializer=lambda v: OrderEvent(**json.loads(v.decode("utf-8")))
)

for msg in consumer:
    order: OrderEvent = msg.value
    print(f"Order #{order.order_id}: Rp{order.total:,.0f} — {order.status}")
    consumer.commit()

Topic Management #

from kafka.admin import KafkaAdminClient, NewTopic
from kafka.errors import TopicAlreadyExistsError

def buat_topic(
    nama:         str,
    num_partitions: int = 3,
    replication_factor: int = 1
) -> None:
    admin = KafkaAdminClient(bootstrap_servers=BOOTSTRAP_SERVERS)

    topic = NewTopic(
        name=nama,
        num_partitions=num_partitions,
        replication_factor=replication_factor,
        topic_configs={
            "retention.ms":     str(7 * 24 * 60 * 60 * 1000),  # keep for 7 days
            "cleanup.policy":   "delete",
            "compression.type": "gzip"
        }
    )

    try:
        admin.create_topics([topic])
        print(f"Topic '{nama}' created successfully with {num_partitions} partitions.")
    except TopicAlreadyExistsError:
        print(f"Topic '{nama}' already exists.")
    finally:
        admin.close()

def list_topics() -> list[str]:
    admin = KafkaAdminClient(bootstrap_servers=BOOTSTRAP_SERVERS)
    topics = admin.list_topics()
    admin.close()
    return [t for t in topics if not t.startswith("__")]  # filter internal topics

def hapus_topic(nama: str) -> None:
    admin = KafkaAdminClient(bootstrap_servers=BOOTSTRAP_SERVERS)
    admin.delete_topics([nama])
    admin.close()
    print(f"Topic '{nama}' deleted.")

# Create topics for the application
buat_topic("orders",       num_partitions=3)
buat_topic("order-events", num_partitions=3)
buat_topic("user-events",  num_partitions=6)
print("Topics:", list_topics())

Error Handling and Dead Letter Queues #

from kafka import KafkaProducer, KafkaConsumer
import json

DLQ_TOPIC = "orders-dlq"   # Dead Letter Queue -- messages that failed processing

def proses_dengan_retry(
    consumer: KafkaConsumer,
    dlq_producer: KafkaProducer,
    maks_retry: int = 3
) -> None:

    for msg in consumer:
        retry_count = 0
        berhasil    = False

        while retry_count <= maks_retry and not berhasil:
            try:
                # Process the message
                proses_order(msg.value)
                consumer.commit()
                berhasil = True

            except Exception as e:
                retry_count += 1
                print(f"Retry {retry_count}/{maks_retry} for offset {msg.offset}: {e}")

                if retry_count > maks_retry:
                    # Send to the Dead Letter Queue
                    dlq_payload = {
                        "original_topic":     msg.topic,
                        "original_partition": msg.partition,
                        "original_offset":    msg.offset,
                        "original_key":       msg.key,
                        "original_value":     msg.value,
                        "error":              str(e),
                        "failed_at":          datetime.utcnow().isoformat()
                    }
                    dlq_producer.send(DLQ_TOPIC, value=dlq_payload)
                    dlq_producer.flush()
                    consumer.commit()  # commit so it doesn't loop forever
                    print(f"Message sent to DLQ: {DLQ_TOPIC}")

When to Choose Kafka vs Other Brokers #

Choose Kafka when:
  ✓ Very high throughput (millions of messages/second)
  ✓ You need message replay (consumers can re-read from any offset)
  ✓ Many independent consumers read the same topic
  ✓ Event sourcing or long-term audit logs
  ✓ Real-time data pipelines across many services

Choose RabbitMQ when:
  ✓ Complex message routing (exchanges, bindings, routing keys)
  ✓ You need more granular per-message acknowledgments
  ✓ Small cluster size and a team unfamiliar with Kafka
  ✓ Message TTL and priority queue needs

Choose Amazon SQS / Google Pub/Sub when:
  ✓ You want a managed service without cluster operations
  ✓ You're already in the AWS / GCP ecosystem

Summary #

enable_auto_commit=False — always use manual commits so messages aren’t considered done before they’re actually processed; auto commit can cause message loss if it crashes during processing.
The key determines the partition — messages with the same key always go to the same partition, preserving order per entity (e.g., per user_id or order_id).
acks="all" — use it in the producer to ensure messages are stored on all replicas before being considered successful; trades latency for durability.
auto_offset_reset — "earliest" to read from the beginning (new consumer/group); "latest" to only read new messages after the consumer is active.
poll() with a timeout — use consumer.poll(timeout_ms=1000) instead of direct for msg in consumer iteration so you can check the shutdown flag periodically.
Consumer Groups for scalability — run several consumer instances with the same group_id; Kafka automatically distributes partitions across instances.
Explicit serialization — always define value_serializer and value_deserializer in the constructor; don’t send raw bytes without a defined format.
Dead Letter Queues (DLQ) — send messages that fail processing after N retries to a separate topic so they don’t block other message processing.
KRaft mode — use Kafka 3.x with KRaft mode for new installations; no more Zookeeper needed.
producer.flush() before exiting — make sure all buffered messages are sent before the application stops.

← Previous: Elasticsearch Next: RabbitMQ →