Kafka Streams & Patterns

Input topic "user-updates" (key = user-id):
  offset 0: {key: "u1", value: {name: "Alice", city: "NYC"}}
  offset 1: {key: "u2", value: {name: "Bob", city: "LA"}}
  offset 2: {key: "u1", value: {name: "Alice", city: "SF"}}  ← Alice moved

As KStream (event stream):
  Record 0: u1 → {name: "Alice", city: "NYC"}   ← event: Alice was in NYC
  Record 1: u2 → {name: "Bob", city: "LA"}      ← event: Bob was in LA
  Record 2: u1 → {name: "Alice", city: "SF"}    ← event: Alice moved to SF
  All 3 records exist and matter.

As KTable (changelog / current state):
  u1 → {name: "Alice", city: "SF"}   ← latest value for u1
  u2 → {name: "Bob", city: "LA"}     ← latest value for u2
  Only 2 entries — u1's old value is replaced.

Operations:
  KStream: filter, map, flatMap, branch, merge, join (windowed)
  KTable: filter, mapValues, join (non-windowed), aggregate
  KStream → KTable: groupByKey().aggregate() (stateful transformation)
  KTable → KStream: toStream() (emit changes as events)

Topic: "orders.events" (compacted for snapshots, or delete for full history)

Events for order-123:
  offset 0: {key: "order-123", event: "OrderCreated", items: [...], total: 99}
  offset 1: {key: "order-123", event: "PaymentReceived", amount: 99}
  offset 2: {key: "order-123", event: "OrderShipped", trackingId: "TRK-456"}
  offset 3: {key: "order-123", event: "OrderDelivered", timestamp: "..."}

Rebuilding current state:
  Start with empty state {}
  Apply OrderCreated    → {status: "created", items: [...], total: 99}
  Apply PaymentReceived → {status: "paid", items: [...], total: 99}
  Apply OrderShipped    → {status: "shipped", trackingId: "TRK-456"}
  Apply OrderDelivered  → {status: "delivered", deliveredAt: "..."}

Benefits:
  ✓ Complete audit trail — every change is recorded
  ✓ Time travel — rebuild state at any point in time
  ✓ Multiple projections — different views from same events
  ✓ Debugging — replay events to reproduce bugs

Trade-offs:
  ✗ Storage grows with every event (mitigate with snapshots)
  ✗ Replay time increases over time (mitigate with periodic snapshots)
  ✗ Schema evolution is harder (events are immutable, schemas change)

Architecture:

  ┌─────────────┐     ┌─────────────┐     ┌─────────────────────┐
  │   Command   │     │    Kafka    │     │    Query Service     │
  │   Service   │────▶│   Topics    │────▶│  (Read Projections)  │
  │  (Writes)   │     │             │     │                      │
  └─────────────┘     └─────────────┘     └─────────────────────┘
        │                    │                       │
   Validates &          Durable event           Builds optimized
   produces events      log (source             read models:
   to Kafka             of truth)               - Elasticsearch
                                                - Redis cache
                                                - Denormalized DB

Example — E-commerce:
  Command side:
    POST /orders → validates → produces "OrderCreated" to Kafka
    POST /payments → validates → produces "PaymentProcessed" to Kafka

  Query side (multiple independent consumers):
    Consumer 1: Builds order-status view in Redis (fast lookups)
    Consumer 2: Builds search index in Elasticsearch (full-text search)
    Consumer 3: Builds analytics aggregates in ClickHouse (reporting)
    Consumer 4: Sends notification emails (side effect)

Benefits:
  ✓ Read and write models scale independently
  ✓ Read models optimized for specific query patterns
  ✓ Multiple read models from same event stream
  ✓ Write side is simple — just validate and produce events

Topic: "orders.created"

Producer: Order Service publishes OrderCreated event (once)

Consumer Group "inventory":    → Reserves stock
Consumer Group "payments":     → Initiates payment charge
Consumer Group "notifications": → Sends confirmation email
Consumer Group "analytics":    → Updates real-time dashboard
Consumer Group "fraud":        → Runs fraud detection rules

Each group:
  - Gets ALL records independently
  - Has its own offsets (different processing speeds)
  - Can be scaled independently
  - Can be added/removed without affecting others

Why this beats point-to-point:
  - Order Service doesn't know about downstream consumers
  - Adding a new consumer requires zero changes to the producer
  - Each consumer can fail independently without affecting others

Happy path:
  1. Order Service → publishes "OrderCreated" to orders.events
  2. Payment Service (consumes OrderCreated) → charges card
     → publishes "PaymentSucceeded" to payments.events
  3. Inventory Service (consumes PaymentSucceeded) → reserves stock
     → publishes "StockReserved" to inventory.events
  4. Shipping Service (consumes StockReserved) → creates shipment
     → publishes "ShipmentCreated" to shipping.events
  5. Order Service (consumes ShipmentCreated) → updates order status

Compensation (failure path):
  3. Inventory Service → stock unavailable
     → publishes "StockReservationFailed" to inventory.events
  4. Payment Service (consumes StockReservationFailed) → refunds payment
     → publishes "PaymentRefunded" to payments.events
  5. Order Service (consumes PaymentRefunded) → marks order as cancelled

Challenges:
  - Debugging distributed flows is hard (use correlation IDs)
  - No central coordinator — harder to see the full picture
  - Compensating transactions must be idempotent
  - Monitoring requires distributed tracing (OpenTelemetry)

Consumer processing flow:

  poll() → records
  for each record:
    try:
      process(record)           // Happy path
    catch (RetriableException):
      retry up to 3 times       // Transient failure (network, timeout)
    catch (NonRetriableException):
      publish to DLQ topic      // Permanent failure (bad data, schema mismatch)
      commit offset             // Move past the poison pill
      alert operations team

DLQ topic naming: "<original-topic>.dlq"
  orders.created → orders.created.dlq

DLQ record should include:
  - Original record (key, value, headers)
  - Error message and stack trace
  - Original topic, partition, offset
  - Timestamp of failure
  - Retry count

DLQ processing strategies:
  1. Manual inspection — ops team reviews and fixes
  2. Automated retry — separate consumer retries DLQ after delay
  3. Alerting — trigger PagerDuty/Slack on DLQ growth
  4. Replay — fix the bug, then replay DLQ back to original topic

Architecture:

  ┌──────────────┐     ┌───────────────┐     ┌─────────────┐
  │  PostgreSQL  │     │   Debezium    │     │    Kafka    │
  │   (source)   │────▶│  (connector)  │────▶│   Topics    │
  │              │     │               │     │             │
  └──────────────┘     └───────────────┘     └─────────────┘
        │                                          │
   WAL (Write-Ahead                          Downstream consumers:
   Log) captures                             - Elasticsearch (search sync)
   every change                              - Redis (cache invalidation)
                                             - Data warehouse (analytics)
                                             - Another microservice

How Debezium works:
  1. Connects to PostgreSQL's WAL (Write-Ahead Log)
  2. Reads every INSERT/UPDATE/DELETE as it happens
  3. Converts each change to a Kafka record:
     - Key: primary key of the row
     - Value: {before: {...}, after: {...}, op: "u", ts_ms: ...}
  4. Publishes to topic: "dbserver.schema.table"

CDC record structure:
  {
    "before": {"id": 1, "name": "Alice", "email": "old@email.com"},
    "after":  {"id": 1, "name": "Alice", "email": "new@email.com"},
    "op": "u",        // c=create, u=update, d=delete, r=read (snapshot)
    "ts_ms": 1706745600000,
    "source": {"db": "orders", "table": "users", "lsn": 12345}
  }