Apache Kafka: Topics, Producers, and Consumers

Apache Kafka: The Distributed Event Streaming Platform

Apache Kafka is a distributed event streaming platform capable of handling trillions of events per day. Originally developed at LinkedIn in 2011, Kafka has evolved from a messaging queue into a comprehensive streaming platform that serves as the central nervous system for modern data architectures.

Why Kafka Dominates Event Streaming

Core Abstraction:

• Distributed commit log — append-only, partitioned, replicated
• Durable, ordered, replayable event storage

Architectural Advantages:

1. Disk-sequential I/O — high write throughput
2. Zero-copy transfer — efficient consumer reads
3. Partition-level parallelism — horizontal scalability

Kafka vs Traditional Message Brokers:

Key Insight: Unlike traditional message brokers, Kafka retains messages for configurable retention periods, allowing consumers to replay events at any point in time. This enables event sourcing, CQRS, and exactly-once processing patterns.

Architecture Diagram

Key Concepts

Production Code

Kafka Producer with Idempotent Writes

from kafka import KafkaProducer
from kafka.errors import KafkaError
import json
import logging
import uuid
from typing import Dict, Optional
from dataclasses import dataclass, asdict

logger = logging.getLogger(__name__)


@dataclass
class KafkaEvent:
    """Base event class with idempotency support."""
    event_id: str
    event_type: str
    payload: Dict
    timestamp: str

    def to_json(self) -> bytes:
        return json.dumps(asdict(self)).encode("utf-8")


class IdempotentProducer:
    """Kafka producer with idempotent writes, retries, and dead letter handling."""

    def __init__(
        self,
        bootstrap_servers: list,
        topic: str,
        max_retries: int = 3,
        retry_backoff_ms: int = 100,
    ):
        self.topic = topic
        self.producer = KafkaProducer(
            bootstrap_servers=bootstrap_servers,
            acks="all",                    # Wait for all ISR replicas
            enable_idempotence=True,        # Exactly-once semantics
            max_in_flight_requests_per_connection=5,  # Safe with idempotence
            retries=max_retries,
            retry_backoff=retry_backoff_ms,
            compression_type="snappy",     # Reduce network I/O
            linger_ms=10,                  # Batch for throughput
            batch_size=32768,              # 32KB batches
            buffer_memory=67108864,        # 64MB buffer
        )
        self.dlq_topic = f"{topic}.dlq"

    def send(
        self,
        event: KafkaEvent,
        key: Optional[str] = None,
        callback=None,
    ) -> None:
        """Send an event to Kafka with idempotency and error handling."""
        try:
            future = self.producer.send(
                topic=self.topic,
                key=key.encode("utf-8") if key else None,
                value=event.to_json(),
                headers=[("event_type", event.event_type.encode())],
            )
            if callback:
                future.add_callback(callback)
            future.add_errback(self._handle_error, event)
        except Exception as e:
            logger.error(f"Failed to produce event {event.event_id}: {e}")
            self._send_to_dlq(event, str(e))

    def _handle_error(self, exc: KafkaError, event: KafkaEvent) -> None:
        """Handle producer errors and route to dead letter queue."""
        logger.error(f"Error producing event {event.event_id}: {exc}")
        self._send_to_dlq(event, str(exc))

    def _send_to_dlq(self, event: KafkaEvent, error: str) -> None:
        """Route failed events to dead letter queue."""
        dlq_event = KafkaEvent(
            event_id=str(uuid.uuid4()),
            event_type="dead_letter",
            payload={
                "original_event": asdict(event),
                "error": error,
            },
            timestamp=event.timestamp,
        )
        self.producer.send(
            topic=self.dlq_topic,
            value=dlq_event.to_json(),
        )
        logger.warning(f"Event {event.event_id} routed to DLQ")

    def flush(self) -> None:
        """Ensure all pending messages are sent."""
        self.producer.flush(timeout=30)

    def close(self) -> None:
        """Gracefully shut down the producer."""
        self.producer.flush(timeout=30)
        self.producer.close()


# Usage
producer = IdempotentProducer(
    bootstrap_servers=["kafka-broker-1:9092", "kafka-broker-2:9092"],
    topic="orders.events",
)

event = KafkaEvent(
    event_id=str(uuid.uuid4()),
    event_type="order_created",
    payload={"order_id": "ORD-12345", "amount": 99.99, "currency": "USD"},
    timestamp="2024-01-15T10:30:00Z",
)
producer.send(event, key="ORD-12345", callback=lambda metadata: logger.info(
    f"Sent to {metadata.topic}[{metadata.partition}]@{metadata.offset}"
))
producer.flush()
producer.close()

Kafka Consumer with Offset Management

from kafka import KafkaConsumer
from kafka.errors import KafkaError
import json
import logging
import signal
import sys
from typing import Callable, Dict

logger = logging.getLogger(__name__)


class ReliableConsumer:
    """Kafka consumer with graceful shutdown, offset management, and error handling."""

    def __init__(
        self,
        bootstrap_servers: list,
        topic: str,
        group_id: str,
        handler: Callable[[Dict], None],
        auto_offset_reset: str = "earliest",
        max_poll_records: int = 500,
    ):
        self.handler = handler
        self.running = True
        self.consumer = KafkaConsumer(
            topic,
            bootstrap_servers=bootstrap_servers,
            group_id=group_id,
            auto_offset_reset=auto_offset_reset,
            enable_auto_commit=False,  # Manual commit for reliability
            max_poll_records=max_poll_records,
            session_timeout_ms=30000,
            heartbeat_interval_ms=10000,
            max_poll_interval_ms=300000,
            value_deserializer=lambda m: json.loads(m.decode("utf-8")),
            key_deserializer=lambda k: k.decode("utf-8") if k else None,
        )

        # Register signal handlers for graceful shutdown
        signal.signal(signal.SIGINT, self._shutdown)
        signal.signal(signal.SIGTERM, self._shutdown)

    def _shutdown(self, signum, frame):
        """Handle shutdown signals gracefully."""
        logger.info("Shutdown signal received, closing consumer...")
        self.running = False

    def consume(self) -> None:
        """Main consumption loop with manual offset commits."""
        logger.info("Starting consumption loop...")
        while self.running:
            try:
                records = self.consumer.poll(timeout_ms=1000)
                for topic_partition, messages in records.items():
                    for message in messages:
                        try:
                            self.handler(message.value)
                        except Exception as e:
                            logger.error(
                                f"Error processing message at "
                                f"{message.topic}[{message.partition}]@{message.offset}: {e}"
                            )
                            # Don't commit — message will be reprocessed on restart
                            continue

                # Commit after successful processing
                if records:
                    self.consumer.commit()
                    logger.debug(f"Committed offsets for {len(records)} partitions")

            except KafkaError as e:
                logger.error(f"Kafka error during polling: {e}")
                continue

        self.consumer.close()
        logger.info("Consumer closed gracefully")

    def resume(self) -> None:
        """Resume consumption after a pause."""
        self.running = True


# Usage
def process_order(event: dict) -> None:
    """Process a single order event."""
    logger.info(f"Processing order: {event.get('order_id')}")
    # Business logic here
    # If processing fails, an exception will prevent offset commit


consumer = ReliableConsumer(
    bootstrap_servers=["kafka-broker-1:9092", "kafka-broker-2:9092"],
    topic="orders.events",
    group_id="order-processing-group",
    handler=process_order,
    auto_offset_reset="earliest",
    max_poll_records=500,
)
consumer.consume()

Best Practices

1. Set acks=all with min.insync.replicas=2 in production to ensure writes survive broker failures.
2. Enable idempotent producers (enable_idempotence=True) to prevent duplicate messages on retries.
3. Use manual offset commits (enable_auto_commit=False) to ensure messages are only acknowledged after successful processing.
4. Monitor consumer lag continuously. Alert when lag exceeds the message retention period.
5. Choose partition count carefully — you can increase but never decrease partitions without data migration.
6. Use schema registry (Confluent or Apicurio) for schema evolution. Prefer Avro or Protobuf over JSON for production topics.
7. Implement dead letter queues for messages that fail processing. Never silently drop failed messages.
8. Set max.poll.records appropriately to prevent poll loops from exceeding max.poll.interval.ms.
9. Use static group membership (group.instance.id) to reduce rebalance frequency during temporary disconnections.
10. Document topic ownership — every topic should have a designated team responsible for schema changes and capacity planning.

Kafka Cluster Sizing Reference

See Also

• 020 - Kafka Streams: DSL, Windowed Aggregation, and Exactly-Once - Stream processing with Kafka
• 022 - Spark Structured Streaming: Triggers, Watermarks, and State Management - Alternative stream processing engine
• 024 - Data Ingestion Patterns: Batch, Streaming, CDC, and APIs - Kafka as ingestion backbone
• 023 - Batch vs Streaming: Lambda, Kappa, and Architecture Trade-offs - Streaming architecture patterns
• 030 - Capstone Project: Real-Time Streaming Pipeline - End-to-end Kafka project