yigityildiz0 Event-Driven Architecture
Comprehensive guidance on building event-driven systems, covering event sourcing, CQRS, message broker selection (Kafka, RabbitMQ, NATS), event schema design with CloudEvents, saga patterns, idempotency, and strategies for event versioning and evolution.
When to Use This Skill
Use this skill for:
- Designing publish/subscribe or event streaming systems
- Implementing event sourcing with an event store
- Applying CQRS to separate read and write models
- Choosing between Kafka, RabbitMQ, AWS SNS/SQS, and NATS
- Implementing the saga pattern (orchestration or choreography)
- Designing event schemas with CloudEvents, Avro, or Protobuf
- Handling idempotency and exactly-once semantics
- Managing dead letter queues and event replay
- Versioning events without breaking consumers
Trigger phrases: "event-driven", "event sourcing", "CQRS", "message broker", "Kafka", "RabbitMQ", "pub/sub", "saga pattern", "event store", "dead letter queue", "event schema", "CloudEvents", "eventual consistency"
What This Skill Does
Provides production-ready event-driven patterns including:
- Core Patterns: Pub/sub, event streaming, event sourcing, CQRS
- Broker Selection: Feature comparison and trade-offs for Kafka, RabbitMQ, NATS, SNS/SQS
- Schema Design: CloudEvents format, Avro schemas, schema registry, event versioning
- Reliability: Idempotency, outbox pattern, dead letter queues, exactly-once semantics
- Orchestration: Saga pattern (orchestrator-based and choreography-based)
- Operations: Consumer groups, partitioning, replay, monitoring
Instructions
Step 1: Understand Event-Driven Patterns
┌────────────────────────────────────────────────────────────────────┐
│ EVENT-DRIVEN PATTERNS │
├──────────────────┬──────────────────┬──────────────────────────────┤
│ PUB/SUB │ EVENT STREAMING │ EVENT SOURCING │
│ │ │ │
│ Fire-and-forget │ Ordered log of │ Persist all state changes │
│ message │ events with │ as events; derive current │
│ broadcasting │ consumer replay │ state by replaying │
│ │ │ │
│ Use: notifs, │ Use: data │ Use: audit, undo, temporal │
│ webhooks, │ pipelines, │ queries, complex domains │
│ decoupling │ CDC, analytics │ │
│ │ │ │
│ Tools: │ Tools: │ Tools: │
│ RabbitMQ, SNS, │ Kafka, Pulsar, │ EventStoreDB, Kafka + │
│ Redis Pub/Sub │ Kinesis, NATS │ custom store, Axon │
│ │ JetStream │ │
└──────────────────┴──────────────────┴──────────────────────────────┘
Pattern Selection Guide:
Q: Do I need to replay past events?
├── No -> Simple pub/sub (RabbitMQ, SNS/SQS)
└── Yes
Q: Do I need to reconstruct state from events?
├── No -> Event streaming (Kafka, NATS JetStream)
└── Yes -> Event sourcing (EventStoreDB, custom event store)
Q: Are reads and writes significantly different in shape or scale?
├── No -> Single model is fine
└── Yes -> CQRS (separate read/write models, eventually consistent)
Step 2: Choose a Message Broker
Broker Comparison:
┌──────────────┬────────────────┬─────────────────┬───────────────┬──────────────┐
│ Feature │ Apache Kafka │ RabbitMQ │ AWS SNS/SQS │ NATS │
├──────────────┼────────────────┼─────────────────┼───────────────┼──────────────┤
│ Model │ Log-based │ Queue/exchange │ Queue + topic │ Pub/sub + │
│ │ streaming │ broker │ cloud managed │ JetStream │
├──────────────┼────────────────┼─────────────────┼───────────────┼──────────────┤
│ Ordering │ Per-partition │ Per-queue │ FIFO optional │ Per-subject │
│ │ guaranteed │ guaranteed │ (SQS FIFO) │ (JetStream) │
├──────────────┼────────────────┼─────────────────┼───────────────┼──────────────┤
│ Retention │ Configurable │ Until consumed │ Up to 14 days │ Configurable │
│ │ (days/size) │ │ (SQS) │ (JetStream) │
├──────────────┼────────────────┼─────────────────┼───────────────┼──────────────┤
│ Replay │ Yes (offset) │ No (dead letter │ No (DLQ only) │ Yes │
│ │ │ only) │ │ (JetStream) │
├──────────────┼────────────────┼─────────────────┼───────────────┼──────────────┤
│ Throughput │ Very high │ Moderate │ High │ Very high │
│ │ (millions/sec) │ (tens of k/sec) │ (managed) │ (millions/s) │
├──────────────┼────────────────┼─────────────────┼───────────────┼──────────────┤
│ Complexity │ High │ Moderate │ Low │ Low-Moderate │
├──────────────┼────────────────┼─────────────────┼───────────────┼──────────────┤
│ Best For │ Event streams, │ Task queues, │ Serverless, │ Microservice │
│ │ CDC, analytics │ RPC, routing │ AWS-native │ messaging │
└──────────────┴────────────────┴─────────────────┴───────────────┴──────────────┘
Step 3: Design Event Schemas
CloudEvents Format (JSON):
{
"specversion": "1.0",
"id": "evt-a1b2c3d4-e5f6-7890-abcd-ef1234567890",
"source": "urn:example:order-service",
"type": "com.example.order.created",
"datacontenttype": "application/json",
"time": "2026-03-03T10:30:00Z",
"subject": "order-12345",
"data": {
"orderId": "order-12345",
"customerId": "cust-67890",
"items": [
{
"productId": "prod-111",
"quantity": 2,
"unitPrice": 29.99
}
],
"total": 59.98,
"currency": "USD"
}
}
Event Naming Convention:
Format: <domain>.<aggregate>.<past-tense-verb>
Examples:
com.example.order.created
com.example.order.cancelled
com.example.payment.processed
com.example.inventory.reserved
com.example.shipment.dispatched
com.example.user.email_verified
Rules:
1. Always past tense (something that HAS happened)
2. Domain-qualified to avoid collisions
3. Specific enough to convey meaning without reading the payload
Avro Schema (for schema registry):
{
"type": "record",
"name": "OrderCreated",
"namespace": "com.example.order.events",
"fields": [
{"name": "orderId", "type": "string"},
{"name": "customerId", "type": "string"},
{"name": "items", "type": {
"type": "array",
"items": {
"type": "record",
"name": "OrderItem",
"fields": [
{"name": "productId", "type": "string"},
{"name": "quantity", "type": "int"},
{"name": "unitPrice", "type": {"type": "bytes", "logicalType": "decimal", "precision": 10, "scale": 2}}
]
}
}},
{"name": "total", "type": {"type": "bytes", "logicalType": "decimal", "precision": 10, "scale": 2}},
{"name": "currency", "type": "string", "default": "USD"},
{"name": "occurredAt", "type": {"type": "long", "logicalType": "timestamp-millis"}}
]
}
Step 4: Implement Kafka Producer and Consumer
Python Producer (confluent-kafka):
import json
import uuid
from datetime import datetime, timezone
from confluent_kafka import Producer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer
def create_producer(bootstrap_servers: str) -> Producer:
config = {
"bootstrap.servers": bootstrap_servers,
"acks": "all", # Wait for all replicas
"enable.idempotence": True, # Exactly-once producer semantics
"max.in.flight.requests.per.connection": 5,
"retries": 2147483647, # Infinite retries with idempotence
"linger.ms": 5, # Batch for 5ms for throughput
"compression.type": "lz4", # Compress batches
}
return Producer(config)
def publish_order_created(producer: Producer, order: dict):
"""Publish an OrderCreated event with CloudEvents envelope."""
event = {
"specversion": "1.0",
"id": str(uuid.uuid4()),
"source": "urn:example:order-service",
"type": "com.example.order.created",
"datacontenttype": "application/json",
"time": datetime.now(timezone.utc).isoformat(),
"subject": order["id"],
"data": order,
}
producer.produce(
topic="orders",
key=order["id"].encode("utf-8"), # Partition by order ID
value=json.dumps(event).encode("utf-8"),
headers={
"ce-type": "com.example.order.created",
"ce-source": "urn:example:order-service",
},
callback=delivery_callback,
)
producer.flush()
def delivery_callback(err, msg):
if err:
print(f"Delivery failed: {err}")
else:
print(f"Delivered to {msg.topic()} [{msg.partition()}] @ offset {msg.offset()}")
Python Consumer with Idempotency:
import json
from confluent_kafka import Consumer, KafkaError
from typing import Callable
def create_consumer(
bootstrap_servers: str,
group_id: str,
topics: list[str],
) -> Consumer:
config = {
"bootstrap.servers": bootstrap_servers,
"group.id": group_id,
"auto.offset.reset": "earliest",
"enable.auto.commit": False, # Manual commit for reliability
"max.poll.interval.ms": 300000, # 5 min processing budget
"session.timeout.ms": 45000,
}
consumer = Consumer(config)
consumer.subscribe(topics)
return consumer
class IdempotentConsumer:
"""Consumer that tracks processed event IDs to ensure exactly-once handling."""
def __init__(self, consumer: Consumer, db, handler: Callable):
self.consumer = consumer
self.db = db
self.handler = handler
async def run(self):
try:
while True:
msg = self.consumer.poll(timeout=1.0)
if msg is None:
continue
if msg.error():
if msg.error().code() == KafkaError._PARTITION_EOF:
continue
raise Exception(f"Consumer error: {msg.error()}")
event = json.loads(msg.value().decode("utf-8"))
event_id = event["id"]
# Idempotency check: skip if already processed
if await self.db.execute(
"SELECT 1 FROM processed_events WHERE event_id = :id",
{"id": event_id},
):
self.consumer.commit(msg)
continue
# Process event within a transaction
async with self.db.transaction():
await self.handler(event)
await self.db.execute(
"INSERT INTO processed_events (event_id, processed_at) "
"VALUES (:id, NOW())",
{"id": event_id},
)
self.consumer.commit(msg)
finally:
self.consumer.close()
Step 5: Implement Event Sourcing
Event Store Schema (PostgreSQL):
-- Event store table
CREATE TABLE event_store (
id BIGSERIAL PRIMARY KEY,
aggregate_type TEXT NOT NULL,
aggregate_id UUID NOT NULL,
event_type TEXT NOT NULL,
event_version INTEGER NOT NULL,
data JSONB NOT NULL,
metadata JSONB NOT NULL DEFAULT '{}',
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
-- Optimistic concurrency: no two events for the same aggregate
-- can have the same version
UNIQUE (aggregate_id, event_version)
);
CREATE INDEX idx_event_store_aggregate
ON event_store (aggregate_type, aggregate_id, event_version);
CREATE INDEX idx_event_store_type
ON event_store (event_type, created_at);
-- Processed events table (for idempotent consumers)
CREATE TABLE processed_events (
event_id TEXT PRIMARY KEY,
processed_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
-- Snapshots table (for performance optimization)
CREATE TABLE snapshots (
aggregate_type TEXT NOT NULL,
aggregate_id UUID NOT NULL,
version INTEGER NOT NULL,
state JSONB NOT NULL,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
PRIMARY KEY (aggregate_type, aggregate_id)
);
Event Store Implementation (Python):
import json
import uuid
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import list, Optional
@dataclass
class DomainEvent:
event_type: str
aggregate_id: str
data: dict
version: int = 0
event_id: str = field(default_factory=lambda: str(uuid.uuid4()))
timestamp: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
metadata: dict = field(default_factory=dict)
class EventStore:
def __init__(self, db):
self.db = db
async def append(
self,
aggregate_type: str,
aggregate_id: str,
events: list[DomainEvent],
expected_version: int,
):
"""Append events with optimistic concurrency control."""
async with self.db.transaction():
# Check current version
row = await self.db.fetchone(
"SELECT COALESCE(MAX(event_version), 0) AS version "
"FROM event_store WHERE aggregate_id = :id",
{"id": aggregate_id},
)
current_version = row["version"]
if current_version != expected_version:
raise ConcurrencyError(
f"Expected version {expected_version}, "
f"but current version is {current_version}"
)
for i, event in enumerate(events):
version = expected_version + i + 1
await self.db.execute(
"""INSERT INTO event_store
(aggregate_type, aggregate_id, event_type,
event_version, data, metadata)
VALUES (:agg_type, :agg_id, :evt_type,
:version, :data, :metadata)""",
{
"agg_type": aggregate_type,
"agg_id": aggregate_id,
"evt_type": event.event_type,
"version": version,
"data": json.dumps(event.data),
"metadata": json.dumps(event.metadata),
},
)
async def load(
self,
aggregate_id: str,
after_version: int = 0,
) -> list[DomainEvent]:
"""Load events for an aggregate, optionally after a snapshot version."""
rows = await self.db.fetchall(
"""SELECT event_type, aggregate_id, event_version, data, metadata, created_at
FROM event_store
WHERE aggregate_id = :id AND event_version > :after
ORDER BY event_version""",
{"id": aggregate_id, "after": after_version},
)
return [
DomainEvent(
event_type=r["event_type"],
aggregate_id=r["aggregate_id"],
data=json.loads(r["data"]),
version=r["event_version"],
timestamp=r["created_at"],
metadata=json.loads(r["metadata"]),
)
for r in rows
]
class ConcurrencyError(Exception):
pass
Aggregate Reconstruction:
class OrderAggregate:
"""Reconstruct order state by replaying events."""
def __init__(self):
self.id: Optional[str] = None
self.status: str = "unknown"
self.items: list = []
self.total: float = 0.0
self.version: int = 0
def apply(self, event: DomainEvent):
handler = getattr(self, f"_on_{event.event_type}", None)
if handler:
handler(event.data)
self.version = event.version
def _on_order_created(self, data):
self.id = data["orderId"]
self.status = "pending"
self.items = data["items"]
self.total = data["total"]
def _on_order_confirmed(self, data):
self.status = "confirmed"
def _on_order_cancelled(self, data):
self.status = "cancelled"
@classmethod
async def load(cls, event_store: EventStore, aggregate_id: str) -> "OrderAggregate":
order = cls()
events = await event_store.load(aggregate_id)
for event in events:
order.apply(event)
return order
Step 6: Implement the Saga Pattern
Choreography-Based Saga (event-driven):
Order Service Payment Service Inventory Service Shipping Service
│ │ │ │
│ OrderCreated │ │ │
├──────────────────────►│ │ │
│ │ PaymentProcessed │ │
│ ├──────────────────────►│ │
│ │ │ InventoryReserved │
│ │ ├──────────────────────►│
│ │ │ │
│ │ If PaymentFailed │ │
│ OrderCancelled ◄────┤ │ │
│ │ │ │
│ │ If InsufficientStock│ │
│ │ PaymentRefunded ◄────┤ │
│ OrderCancelled ◄────┤ │ │
Orchestrator-Based Saga (Python):
from enum import Enum
from dataclasses import dataclass
from typing import list, Optional
class SagaStepStatus(Enum):
PENDING = "pending"
COMPLETED = "completed"
COMPENSATING = "compensating"
COMPENSATED = "compensated"
FAILED = "failed"
@dataclass
class SagaStep:
name: str
action: str # Command to execute
compensation: str # Command to undo the action
status: SagaStepStatus = SagaStepStatus.PENDING
class OrderSagaOrchestrator:
"""Orchestrates order creation across multiple services."""
def __init__(self, command_bus, event_store):
self.command_bus = command_bus
self.event_store = event_store
def define_steps(self, order_data: dict) -> list[SagaStep]:
return [
SagaStep(
name="reserve_inventory",
action="ReserveInventory",
compensation="ReleaseInventory",
),
SagaStep(
name="process_payment",
action="ProcessPayment",
compensation="RefundPayment",
),
SagaStep(
name="create_shipment",
action="CreateShipment",
compensation="CancelShipment",
),
]
async def execute(self, saga_id: str, order_data: dict):
steps = self.define_steps(order_data)
completed_steps = []
for step in steps:
try:
step.status = SagaStepStatus.COMPLETED
await self.command_bus.send(step.action, {
"saga_id": saga_id,
"order": order_data,
})
completed_steps.append(step)
except Exception as e:
# Compensate in reverse order
step.status = SagaStepStatus.FAILED
await self._compensate(saga_id, completed_steps, order_data)
raise SagaFailedError(
f"Saga {saga_id} failed at step '{step.name}': {e}"
)
async def _compensate(
self,
saga_id: str,
completed_steps: list[SagaStep],
order_data: dict,
):
for step in reversed(completed_steps):
try:
step.status = SagaStepStatus.COMPENSATING
await self.command_bus.send(step.compensation, {
"saga_id": saga_id,
"order": order_data,
})
step.status = SagaStepStatus.COMPENSATED
except Exception as comp_error:
# Compensation failure requires manual intervention
step.status = SagaStepStatus.FAILED
await self._alert_manual_intervention(saga_id, step, comp_error)
class SagaFailedError(Exception):
pass
Step 7: Handle Event Versioning and Evolution
Versioning Strategies:
# Strategy 1: Upcasting (transform old events to new format on read)
class EventUpcaster:
"""Transform events from older versions to the current schema."""
upcasters = {}
@classmethod
def register(cls, event_type: str, from_version: int, to_version: int):
def decorator(fn):
cls.upcasters[(event_type, from_version, to_version)] = fn
return fn
return decorator
@classmethod
def upcast(cls, event: DomainEvent) -> DomainEvent:
key = (event.event_type, event.metadata.get("schema_version", 1), 2)
upcaster = cls.upcasters.get(key)
if upcaster:
return upcaster(event)
return event
@EventUpcaster.register("order_created", from_version=1, to_version=2)
def upcast_order_created_v1_to_v2(event: DomainEvent) -> DomainEvent:
"""V1 had 'amount'; V2 renamed to 'total' and added 'currency'."""
data = dict(event.data)
data["total"] = data.pop("amount", 0)
data.setdefault("currency", "USD")
event.data = data
event.metadata["schema_version"] = 2
return event
Backward-Compatible Evolution Rules:
SAFE changes (backward compatible):
- Add a new optional field with a default value
- Add a new event type
- Add a new optional header
UNSAFE changes (breaking):
- Remove or rename a field
- Change a field's type
- Change the meaning of a field
- Remove an event type
MIGRATION strategy for breaking changes:
1. Publish both old and new event versions simultaneously
2. Migrate all consumers to read the new version
3. Stop publishing the old version
4. (Optional) Upcast old events on read
Step 8: Implement the Outbox Pattern
# The outbox pattern ensures atomicity between database writes and event publishing.
# Events are written to an outbox table in the same transaction as the business data.
# A separate process reads the outbox and publishes to the message broker.
async def create_order_with_outbox(db, order_data: dict):
"""Atomically persist order and queue event for publishing."""
async with db.transaction():
# 1. Write business data
order = await db.execute(
"INSERT INTO orders (customer_id, total, status) "
"VALUES (:cid, :total, 'pending') RETURNING *",
{"cid": order_data["customer_id"], "total": order_data["total"]},
)
# 2. Write event to outbox (same transaction)
await db.execute(
"""INSERT INTO outbox (aggregate_type, aggregate_id, event_type, payload)
VALUES ('order', :id, 'order.created', :payload)""",
{
"id": order["id"],
"payload": json.dumps({
"orderId": order["id"],
"customerId": order["customer_id"],
"total": order["total"],
}),
},
)
return order
# Outbox publisher (runs as a background process)
async def outbox_publisher(db, producer, poll_interval: float = 1.0):
"""Poll outbox table and publish to Kafka."""
while True:
rows = await db.fetchall(
"SELECT * FROM outbox WHERE published_at IS NULL "
"ORDER BY created_at LIMIT 100",
)
for row in rows:
producer.produce(
topic=row["aggregate_type"] + "s",
key=row["aggregate_id"].encode(),
value=row["payload"].encode(),
)
await db.execute(
"UPDATE outbox SET published_at = NOW() WHERE id = :id",
{"id": row["id"]},
)
producer.flush()
await asyncio.sleep(poll_interval)
RabbitMQ Setup (Docker Compose + Python):
# docker-compose.yaml
services:
rabbitmq:
image: rabbitmq:3.13-management-alpine
ports:
- "5672:5672" # AMQP
- "15672:15672" # Management UI
environment:
RABBITMQ_DEFAULT_USER: admin
RABBITMQ_DEFAULT_PASS: secret
volumes:
- rabbitmq-data:/var/lib/rabbitmq
healthcheck:
test: ["CMD", "rabbitmq-diagnostics", "-q", "ping"]
interval: 10s
timeout: 5s
retries: 5
volumes:
rabbitmq-data:
# RabbitMQ publisher with exchange and routing
import pika
import json
def setup_rabbitmq(host: str = "localhost"):
connection = pika.BlockingConnection(
pika.ConnectionParameters(host=host, credentials=pika.PlainCredentials("admin", "secret"))
)
channel = connection.channel()
# Topic exchange for flexible routing
channel.exchange_declare(exchange="events", exchange_type="topic", durable=True)
# Dead letter exchange
channel.exchange_declare(exchange="events.dlx", exchange_type="topic", durable=True)
# Queues with dead letter routing
channel.queue_declare(
queue="order-processing",
durable=True,
arguments={
"x-dead-letter-exchange": "events.dlx",
"x-dead-letter-routing-key": "order.failed",
"x-message-ttl": 300000, # 5 min TTL
},
)
channel.queue_bind(queue="order-processing", exchange="events", routing_key="order.*")
# Dead letter queue
channel.queue_declare(queue="order-failed-dlq", durable=True)
channel.queue_bind(queue="order-failed-dlq", exchange="events.dlx", routing_key="order.failed")
return connection, channel
def publish_event(channel, routing_key: str, event: dict):
channel.basic_publish(
exchange="events",
routing_key=routing_key,
body=json.dumps(event),
properties=pika.BasicProperties(
delivery_mode=2, # Persistent
content_type="application/json",
message_id=event.get("id", ""),
),
)
Best Practices
- Use the outbox pattern to guarantee atomicity between state changes and event publication
- Design events as facts (past tense, immutable); never update a published event
- Include enough context in events so consumers do not need to call back to the producer
- Implement idempotent consumers using a processed-events table or deduplication key
- Use a schema registry (Confluent, Apicurio) to enforce schema compatibility
- Version events from day one; include a schema_version field in metadata
- Set up dead letter queues for every consumer to capture unprocessable messages
- Partition by aggregate ID so all events for one entity go to the same partition (ordering)
- Monitor consumer lag as a key health metric; alert when lag exceeds a threshold
- Prefer choreography for simple flows (2-3 services); use orchestration for complex sagas (4+)
- Replay events to rebuild read models rather than writing complex migration scripts
Common Patterns
Pattern 1: CQRS with Separate Read Model
# Write side: event-sourced aggregate
async def place_order(command, event_store):
order = await OrderAggregate.load(event_store, command.order_id)
events = order.place(command)
await event_store.append("order", command.order_id, events, order.version)
# Read side: projector updates a denormalized read model
async def on_order_created(event, read_db):
await read_db.execute(
"""INSERT INTO order_view
(id, customer_name, total, status, item_count, created_at)
VALUES (:id, :name, :total, 'pending', :count, :ts)""",
{
"id": event.data["orderId"],
"name": event.data["customerName"],
"total": event.data["total"],
"count": len(event.data["items"]),
"ts": event.timestamp,
},
)
Pattern 2: Dead Letter Queue Handler
async def process_dlq(consumer, alert_service):
"""Process dead letter queue: log, alert, and store for manual review."""
for message in consumer:
await alert_service.notify(
channel="ops",
message=f"DLQ message: {message.topic} | "
f"Key: {message.key} | "
f"Error: {message.headers.get('x-error')}",
)
await store_for_review(message)
consumer.commit(message)
Pattern 3: Event Replay for Read Model Rebuild
async def rebuild_read_model(event_store, projector, aggregate_type: str):
"""Replay all events to rebuild a read model from scratch."""
await projector.truncate() # Clear the read model
offset = 0
batch_size = 1000
while True:
events = await event_store.load_all(
aggregate_type=aggregate_type,
after_id=offset,
limit=batch_size,
)
if not events:
break
for event in events:
await projector.project(event)
offset = event.id
print(f"Rebuilt read model from {offset} events")
Quality Checklist
- Events use past tense and follow a consistent naming convention
- Event schemas include specversion, id, source, type, and timestamp (CloudEvents)
- Outbox pattern ensures atomicity between data writes and event publication
- Consumers are idempotent (duplicate event processing is safe)
- Dead letter queues configured for every consumer
- Schema versioning strategy documented (upcasting or dual-publish)
- Consumer group lag is monitored and alerted
- Partitioning strategy ensures ordering where needed
- Saga compensations handle failure of compensating actions
- Event retention and compaction policies defined
- Read model rebuild procedure documented and tested
- Schema registry enforces backward compatibility
Related Skills
async-patterns- Concurrency patterns for event processingobservability-setup- Monitoring event-driven systemsgraphql-development- GraphQL subscriptions as an event delivery mechanismcloud-architect- Managed messaging services (SNS/SQS, EventBridge, Pub/Sub)
Version: 1.0.0 Last Updated: March 2026
Iterative Refinement Strategy
This skill is optimized for an iterative approach:
- Execute: Perform the core steps defined above.
- Review: Critically analyze the output (coverage, quality, completeness).
- Refine: If targets aren't met, repeat the specific implementation steps with improved context.
- Loop: Continue until the definition of done is satisfied.