Event Driven

{
  "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": "29.99" }]
  }
}

# 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

# 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

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

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()

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
    }
  },
});

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
}

# 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
});

# 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()

# 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"])

# 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)