event-driven

name	event-driven
description	Use when designing or debugging an event-driven system — choosing Kafka partitioning strategies, implementing the outbox pattern, handling dead-letter queues, ensuring idempotent consumers, or making event sourcing decisions.

Event-Driven Architecture Patterns

Design and implementation patterns for reliable, scalable asynchronous systems using message brokers.

When to Activate

• Decoupling services that currently call each other synchronously
• Designing Kafka topics, partitions, or consumer group topology
• Implementing event schemas with versioning and backward compatibility
• Ensuring exactly-once or at-least-once delivery semantics
• Setting up dead-letter queues and poison message handling
• Implementing the outbox pattern to avoid dual-write problems
• Building event sourcing or CQRS read-model projections

Core Concepts

Broker Comparison

Broker	Retention	Ordering	Throughput	Best For
Kafka	Days–forever (log)	Per partition	Very high	Event streaming, audit log, replay
RabbitMQ	Until consumed	Per queue	High	Task queues, RPC, routing flexibility
AWS SQS	Up to 14 days	None (FIFO available)	High	Simple queues, AWS-native workloads
Redis Streams	Configurable	Per stream	High	Low-latency, simple consumers
GCP Pub/Sub	7 days default	None	Very high	GCP-native, push subscriptions

Delivery Semantics

Semantic	How	Tradeoff
At-least-once	Commit offset only after processing	Duplicates possible; consumer must be idempotent
At-most-once	Commit before processing	No duplicates; messages can be lost
Exactly-once	Transactional producer + idempotent consumer	Highest complexity; use outbox pattern instead

Default to at-least-once with idempotent consumers — it's simpler than exactly-once and eliminates loss.

Event Schema Design

Event Envelope

{
  "event_id": "01HXK4...",
  "event_type": "order.placed",
  "schema_version": "1.0",
  "occurred_at": "2025-06-01T12:00:00Z",
  "aggregate_id": "order-abc-123",
  "aggregate_type": "Order",
  "correlation_id": "req-xyz-456",
  "payload": {
    "customer_id": "cust-789",
    "total": "59.99",
    "currency": "USD",
    "items": [{ "sku": "WIDGET-01", "qty": 2, "unit_price":  
  

Rules:

• event_id: globally unique (ULID or UUID v7) — used for idempotency keys
• occurred_at: when the fact happened, not when published
• correlation_id: propagate from the originating request for distributed tracing
• Monetary values: string decimal, not float

Schema Versioning (Avro / JSON Schema)

# GOOD: backward-compatible change — add optional field with default
{
  "event_type": "order.placed",
  "schema_version": "1.1",   # minor bump — consumers on 1.0 still work
  "payload": {
    "customer_id": "...",
    "total": "...",
    "promo_code": null        # new optional field, defaults to null
  }
}

# BAD: breaking changes
# - removing a field consumers depend on
# - renaming a field
# - changing a field type (string → int)
# → always create a new event type or bump major version

Kafka Patterns

Topic and Partition Design

# Naming: <domain>.<entity>.<event-type> or <domain>.<entity>
orders.placed
orders.cancelled
payments
inventory.stock-updates

# Partition key selection — determines ordering guarantee
producer.produce(topic="orders", key=order.customer_id, ...)  # all orders per customer are ordered
producer.produce(topic="payments", key=payment.order_id, ...)  # all events per order are ordered

# Partition count tradeoffs
low partitions (1–6):   easy rebalancing, less parallelism
high partitions (12+):  more parallelism, more consumer instances, slower rebalance
rule: start with max(consumer_instances * 2, 12), increase later

Python Producer

from confluent_kafka import Producer, KafkaException
import json, uuid
from datetime import datetime, timezone

producer = Producer({
    "bootstrap.servers": "kafka:9092",
    "acks": "all",                    # wait for all ISR replicas
    "retries": 5,
    "retry.backoff.ms": 500,
    "enable.idempotence": True,       # exactly-once producer semantics
    "compression.type": "lz4",
})

def publish_event(topic: str, key: str, event_type: str, payload: dict):
    event = {
        "event_id": str(uuid.uuid4()),
        "event_type": event_type,
        "schema_version": "1.0",
        "occurred_at": datetime.now(timezone.utc).isoformat(),
        "payload": payload,
    }
    producer.produce(
        topic=topic,
        key=key.encode(),
        value=json.dumps(event).encode(),
        on_delivery=_delivery_report,
    )
    producer.poll(0)  # trigger callbacks without blocking

def _delivery_report(err, msg):
    if err:
        logger.error("delivery_failed", topic=msg.topic(), error=str(err))

producer.flush()  # call at shutdown

Python Consumer

from confluent_kafka import Consumer, KafkaError
import json

consumer = Consumer({
    "bootstrap.servers": "kafka:9092",
    "group.id": "notification-service",
    "auto.offset.reset": "earliest",
    "enable.auto.commit": False,      # manual commit after processing
    "max.poll.interval.ms": 300_000,
})

consumer.subscribe(["orders"])

try:
    while True:
        msg = consumer.poll(timeout=1.0)
        if msg is None:
            continue
        if msg.error():
            if msg.error().code() == KafkaError.PARTITION_EOF:
                continue
            raise KafkaException(msg.error())

        event = json.loads(msg.value())
        try:
            handle_event(event)
            consumer.commit(message=msg)   # commit only after success
        except Exception as e:
            logger.error("processing_failed", event_id=event["event_id"], error=str(e))
            # don't commit — message will be redelivered
finally:
    consumer.close()

TypeScript Consumer (KafkaJS)

import { Kafka } from "kafkajs";

const kafka = new Kafka({ brokers: ["kafka:9092"] });
const consumer = kafka.consumer({ groupId: "notification-service" });

await consumer.connect();
await consumer.subscribe({ topics: ["orders"], fromBeginning: false });

await consumer.run({
  autoCommit: false,
  eachMessage: async ({ topic, partition, message, heartbeat }) => {
    const event = JSON.parse(message.value!.toString());
    try {
      await handleEvent(event);
      await consumer.commitOffsets([{
        topic, partition,
        offset: (Number(message.offset) + 1).toString(),
      }]);
    } catch (err) {
      logger.error({ eventId: event.event_id, err }, "processing_failed");
      throw err;  // re-throw to pause and retry
    }
  },
});

Go Consumer (Sarama)

import "github.com/IBM/sarama"

type OrderHandler struct{ db *sql.DB }

func (h *OrderHandler) Setup(_ sarama.ConsumerGroupSession) error   { return nil }
func (h *OrderHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil }

func (h *OrderHandler) ConsumeClaim(sess sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
    for msg := range claim.Messages() {
        var event Event
        if err := json.Unmarshal(msg.Value, &event); err != nil {
            logger.Error("unmarshal_failed", "offset", msg.Offset)
            sess.MarkMessage(msg, "")  // skip unparseable messages
            continue
        }
        if err := h.handle(sess.Context(), event); err != nil {
            return err  // stop processing; session will restart
        }
        sess.MarkMessage(msg, "")
    }
    return nil
}

Idempotency

Consumers must handle duplicate messages — at-least-once delivery guarantees redelivery on failure.

# Pattern: idempotency key in a processed-events table
def handle_order_placed(event: dict, db: Session):
    event_id = event["event_id"]

    # Check if already processed
    if db.query(ProcessedEvent).filter_by(event_id=event_id).first():
        logger.info("duplicate_skipped", event_id=event_id)
        return

    # Process in a transaction that also inserts the idempotency record
    with db.begin():
        create_notification(event["payload"])
        db.add(ProcessedEvent(event_id=event_id, processed_at=datetime.utcnow()))

// Alternative: upsert on natural key (e.g. order_id)
await db.transaction(async (trx) => {
  await trx("notifications")
    .insert({ order_id: event.payload.order_id, customer_id: event.payload.customer_id })
    .onConflict("order_id")
    .ignore();  // safe to call multiple times
});

Outbox Pattern

Solves dual-write: never publish directly from application code after a DB write — they can desync.

# BAD: dual write (either step can fail independently)
BEGIN TRANSACTION
  INSERT INTO orders (...)
COMMIT
kafka.produce("order.placed", ...)   # ← this can fail silently

# GOOD: outbox pattern
BEGIN TRANSACTION
  INSERT INTO orders (...)
  INSERT INTO outbox (event_type, payload, published=false, ...)  # same transaction
COMMIT

# Separate relay process (Debezium CDC or polling relay)
SELECT * FROM outbox WHERE published = false ORDER BY created_at LIMIT 100
→ produce to Kafka
→ UPDATE outbox SET published = true WHERE id IN (...)

# Polling relay (simple, no CDC dependency)
def relay_outbox(db: Session, producer: Producer):
    events = db.query(OutboxEvent).filter_by(published=False).limit(100).all()
    for event in events:
        producer.produce(
            topic=event.topic,
            key=event.aggregate_id.encode(),
            value=event.payload.encode(),
        )
    producer.flush()
    for event in events:
        event.published = True
    db.commit()

Dead-Letter Queues

# Python — send to DLQ after max retries
MAX_RETRIES = 3

def handle_with_dlq(consumer, dlq_producer, msg):
    retry_count = int(msg.headers().get("retry-count", b"0"))
    event = json.loads(msg.value())

    try:
        process_event(event)
        consumer.commit(message=msg)
    except Exception as e:
        if retry_count >= MAX_RETRIES:
            dlq_producer.produce(
                topic=f"{msg.topic()}.dlq",
                key=msg.key(),
                value=msg.value(),
                headers={"original-topic": msg.topic(), "error": str(e)},
            )
            consumer.commit(message=msg)  # move past it
            logger.error("sent_to_dlq", event_id=event["event_id"])
        else:
            # Re-queue with incremented retry count (or let Kafka retry via pause)
            logger.warning("retry", attempt=retry_count + 1, event_id=event["event_id"])

DLQ naming convention: <original-topic>.dlq DLQ review: alert on DLQ lag > 0; investigate and replay or discard manually.

Event Sourcing

Store state as an append-only log of events; derive current state by replaying.

# Events are facts, not commands
EVENTS = [
    {"type": "OrderCreated",   "payload": {"customer_id": "c1", "items": [...]}},
    {"type": "ItemAdded",      "payload": {"sku": "X", "qty": 1}},
    {"type": "OrderConfirmed", "payload": {"confirmed_at": "2025-01-01T..."}},
]

def replay(events: list[dict]) -> Order:
    order = Order()
    for event in events:
        match event["type"]:
            case "OrderCreated":   order.apply_created(event["payload"])
            case "ItemAdded":      order.apply_item_added(event["payload"])
            case "OrderConfirmed": order.apply_confirmed(event["payload"])
    return order

# Snapshot: periodically persist current state to avoid full replay
# Projection: consume event stream to build read models (CQRS read side)

Use Event Sourcing When	Avoid When
Audit trail is a first-class requirement	Simple CRUD with no history needs
Need to replay history for new features	Team unfamiliar with the pattern
Multiple read models from one write model	Strong consistency required across aggregates
Temporal queries ("state at time T")	Simple, low-volume domain

See also: microservices, observability

Red Flags

• Kafka consumer without idempotency — at-least-once delivery means the same message can arrive twice; design consumers to be idempotent before assuming exactly-once semantics
• Hot partition from a low-cardinality key — using event_type as the partition key routes all messages of one type to one partition; choose a high-cardinality key like entity_id for even distribution
• No dead-letter queue for unprocessable messages — a consumer that throws on a bad message blocks all subsequent messages on that partition; route poison messages to a DLQ immediately
• Outbox table without a reliable poller — writing to the outbox without a dedicated transactional poller means events may silently never be published; the poller is half the pattern
• Schema changes without versioning — adding a required field to an event schema breaks all existing consumers silently; always version events and maintain backward compatibility
• Synchronous HTTP calls inside a consumer handler — an upstream timeout blocks the consumer and grows partition lag; use async clients or pre-fetch data outside the consumer loop
• Resetting offsets to earliest on every consumer restart — without committed offsets, a restarted consumer reprocesses all historical events; commit offsets after processing and handle replay explicitly

Checklist

• Event envelope includes event_id, event_type, schema_version, occurred_at, correlation_id
• Partition key chosen to guarantee ordering for events that must be ordered
• Producer uses acks=all and enable.idempotence=true
• Consumer commits offsets manually after successful processing
• All consumers are idempotent (duplicate event_id handled gracefully)
• Outbox pattern used — never dual-write to DB and broker in separate transactions
• Dead-letter queue configured; alerts fire when DLQ lag exceeds threshold
• Event schema changes are backward-compatible (add optional fields only)
• Consumer group IDs are service-specific and stable across deploys
• Dead-letter messages include original topic, offset, and error reason in headers
• Retention period set to cover the longest plausible consumer downtime plus buffer