Design Event-Driven Systems with Event Buses and Brokers

Overview

Synchronous service-to-service calls create tight coupling. The caller has to know where the callee lives, wait for a response, and handle failures itself. When the callee is slow or down, the caller suffers with it. As systems grow, this web of direct dependencies becomes the real problem: a change in one service ripples through every service that calls it.

Event-driven architecture flips this around. Instead of calling each other directly, services drop events on a broker and move on. The OrderPlaced event only needs publishing once. Inventory subscribes and decrements stock. Billing subscribes and issues the invoice. Shipping subscribes and prepares the label. Each one works at its own pace — if billing is slow, orders and shipping keep going. What follows is the practical part — event patterns, broker selection, and working code for Kafka, RabbitMQ, and AWS EventBridge.

When to use it

Use this recipe when:

• Several services must react to the same business event
• Workloads are bursty and need buffering to smooth traffic spikes
• Services have different availability requirements and can’t block on each other
• You need an audit trail where every state change is recorded
• You’re implementing event sourcing for temporal queries and state reconstruction — see CQRS Pattern for read/write separation

Skip it when everything fits in one deployable unit — a broker adds operational overhead that a single service never pays back. It’s also the wrong tool when a request needs an immediate, consistent answer (payment authorization, a stock check at checkout): events settle eventually, not synchronously.

Solution

Publishing Events (Python / Kafka)

from kafka import KafkaProducer
import json

producer = KafkaProducer(
    bootstrap_servers=['kafka:9092'],
    value_serializer=lambda v: json.dumps(v).encode('utf-8'),
    acks='all',
    retries=3,
)

def place_order(order_data):
    # Save order to database
    order = save_order(order_data)

    # Publish event
    event = {
        'type': 'OrderPlaced',
        'aggregate_id': order.id,
        'payload': {
            'customer_id': order.customer_id,
            'items': [item.to_dict() for item in order.items],
            'total': order.total(),
        },
        'occurred_at': order.created_at.isoformat(),
    }

    producer.send('orders', key=order.id.encode(), value=event)
    producer.flush()

    return order

Consuming Events (Node.js / RabbitMQ)

const amqp = require('amqplib');

async function startInventoryConsumer() {
  const connection = await amqp.connect('amqp://rabbitmq');
  const channel = await connection.createChannel();

  const queue = 'inventory_updates';
  await channel.assertQueue(queue, { durable: true });
  await channel.bindQueue(queue, 'orders', 'OrderPlaced');

  channel.consume(queue, async (msg) => {
    if (msg !== null) {
      const event = JSON.parse(msg.content.toString());
      console.log(`Processing ${event.type} for order ${event.aggregate_id}`);

      try {
        await reserveInventory(event.payload.items);
        channel.ack(msg); // Confirm processing
      } catch (error) {
        channel.nack(msg, false, false); // Dead-letter, don't requeue
      }
    }
  });
}

startInventoryConsumer();

AWS EventBridge Event Bus (Terraform)

resource "aws_cloudwatch_event_bus" "main" {
  name = "stackpractices-events"
}

resource "aws_cloudwatch_event_rule" "order_placed" {
  name        = "order-placed-rule"
  event_bus_name = aws_cloudwatch_event_bus.main.name

  event_pattern = jsonencode({
    source      = ["order-service"]
    detail-type = ["OrderPlaced"]
  })
}

resource "aws_cloudwatch_event_target" "inventory_target" {
  rule           = aws_cloudwatch_event_rule.order_placed.name
  event_bus_name = aws_cloudwatch_event_bus.main.name
  arn            = aws_sqs_queue.inventory_queue.arn
}

resource "aws_cloudwatch_event_target" "billing_target" {
  rule           = aws_cloudwatch_event_rule.order_placed.name
  event_bus_name = aws_cloudwatch_event_bus.main.name
  arn            = aws_lambda_function.billing_processor.arn
}

Event Sourcing with Kafka Streams (Java)

public class OrderEventStore {

    private final KafkaStreams streams;

    public OrderEventStore() {
        StreamsBuilder builder = new StreamsBuilder();

        // Event store: aggregate events by order ID
        KStream<String, OrderEvent> eventStream = builder.stream(
            "orders",
            Consumed.with(Serdes.String(), new OrderEventSerde())
        );

        // Materialize current state from event history
        KTable<String, OrderState> orderState = eventStream
            .groupByKey()
            .aggregate(
                OrderState::new,
                (key, event, state) -> state.apply(event),
                Materialized.as("order-state-store")
            );

        // Project to a read model topic
        orderState.toStream().to("order-read-model",
            Produced.with(Serdes.String(), new OrderStateSerde()));

        streams = new KafkaStreams(builder.build(), getStreamsConfig());
    }

    public void start() {
        streams.start();
    }

    public OrderState getOrder(String orderId) {
        ReadOnlyKeyValueStore<String, OrderState> store =
            streams.store(StoreQueryParameters.fromNameAndType(
                "order-state-store",
                QueryableStoreTypes.keyValueStore()
            ));
        return store.get(orderId);
    }
}

// Apply events to reconstruct state
class OrderState {
    private String status;
    private BigDecimal total;
    private List<String> items = new ArrayList<>();

    public OrderState apply(OrderEvent event) {
        switch (event.getType()) {
            case "OrderPlaced":
                this.status = "placed";
                this.total = event.getTotal();
                this.items = event.getItemIds();
                break;
            case "OrderPaid":
                this.status = "paid";
                break;
            case "OrderShipped":
                this.status = "shipped";
                break;
            case "OrderCancelled":
                this.status = "cancelled";
                break;
        }
        return this;
    }
}

Schema Registry with Avro (Python)

from confluent_kafka import Producer, SerializingProducer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer
from dataclasses import dataclass, asdict
import uuid

schema_registry_client = SchemaRegistryClient({
    'url': 'http://schema-registry:8081'
})

order_event_schema_str = """
{
  "type": "record",
  "name": "OrderEvent",
  "namespace": "com.stackpractices.events",
  "fields": [
    {"name": "event_id", "type": "string"},
    {"name": "event_type", "type": "string"},
    {"name": "aggregate_id", "type": "string"},
    {"name": "customer_id", "type": "string"},
    {"name": "total", "type": "double"},
    {"name": "items", "type": {"type": "array", "items": "string"}},
    {"name": "occurred_at", "type": "string"}
  ]
}
"""

avro_serializer = AvroSerializer(
    schema_registry_client,
    order_event_schema_str,
    lambda obj, ctx: asdict(obj)
)

@dataclass
class OrderEvent:
    event_id: str
    event_type: str
    aggregate_id: str
    customer_id: str
    total: float
    items: list
    occurred_at: str

producer = SerializingProducer({
    'bootstrap.servers': 'kafka:9092',
    'value.serializer': avro_serializer,
})

def delivery_report(err, msg):
    # Called once the broker acks (or rejects) the message
    if err is not None:
        print(f"Delivery failed: {err}")

def publish_order_event(order):
    event = OrderEvent(
        event_id=str(uuid.uuid4()),
        event_type='OrderPlaced',
        aggregate_id=order.id,
        customer_id=order.customer_id,
        total=order.total(),
        items=[item.id for item in order.items],
        occurred_at=order.created_at.isoformat(),
    )
    producer.produce(
        topic='orders',
        key=order.id.encode(),
        value=event,
        on_delivery=delivery_report,
    )
    producer.flush()

AWS EventBridge Consumer (TypeScript)

import { EventBridgeClient, PutEventsCommand } from '@aws-sdk/client-eventbridge';
import { SQSClient, ReceiveMessageCommand, DeleteMessageCommand } from '@aws-sdk/client-sqs';

const eventBridge = new EventBridgeClient({ region: 'us-east-1' });
const sqs = new SQSClient({ region: 'us-east-1' });

// Publishing to EventBridge
async function publishOrderEvent(order: Order): Promise<void> {
  const command = new PutEventsCommand({
    Entries: [{
      EventBusName: 'stackpractices-events',
      Source: 'order-service',
      DetailType: 'OrderPlaced',
      Detail: JSON.stringify({
        orderId: order.id,
        customerId: order.customerId,
        total: order.total,
        items: order.items,
      }),
    }],
  });
  await eventBridge.send(command);
}

// Consuming from SQS (EventBridge target)
async function processOrderEvents(): Promise<void> {
  const result = await sqs.send(new ReceiveMessageCommand({
    QueueUrl: process.env.INVENTORY_QUEUE_URL!,
    MaxNumberOfMessages: 10,
    WaitTimeSeconds: 20,
  }));

  for (const message of result.Messages || []) {
    try {
      const event = JSON.parse(message.Body!);
      const detail = JSON.parse(event.detail);

      await reserveInventory(detail.items);
      await sqs.send(new DeleteMessageCommand({
        QueueUrl: process.env.INVENTORY_QUEUE_URL!,
        ReceiptHandle: message.ReceiptHandle!,
      }));
    } catch (error) {
      // Message will become visible again after visibility timeout
      // After max retries, it moves to DLQ
      console.error('Failed to process order event:', error);
    }
  }
}

Explanation

• Event vs command: an event states that something happened (OrderPlaced). It’s immutable and goes to everyone listening. A command asks for an action (PlaceOrder), and one named handler is expected to run it. Don’t mix them — a service that receives a command shouldn’t republish it as an event without transforming it first.
• Message broker patterns: pub/sub fans one event out to every subscriber, while point-to-point hands the event to a single consumer. Point-to-point scales by throwing more workers at the same queue (competing consumers). Pick pub/sub when every service needs the event, point-to-point when only one does.
• Event ordering: brokers don’t guarantee global ordering. If OrderPlaced and OrderCancelled arrive out of sequence, the inventory system may try to cancel stock that was never reserved. Keep ordering inside an aggregate — route every event for order-123 to the same partition — or make handlers idempotent.
• Dead-letter queues: don’t let a message that keeps failing block everything queued behind it. After N retries, route it to a dead-letter queue for manual inspection. Watch DLQ depth as a critical alert — a growing DLQ usually means a systemic problem, not bad luck.

Variants

Broker	Pattern	Durability	Ordering	Scale	Best for
Kafka	Pub/sub, streams	High	Partition-scoped	Massive	Event sourcing, streaming
RabbitMQ	Pub/sub, queues	Medium	Queue-scoped	Medium	Complex routing, AMQP
NATS	Pub/sub, request/reply	Low	None	Very high	Low-latency, simple
AWS SNS/SQS	Pub/sub, queues	High	None	High	Cloud-native, serverless
Redis Streams	Pub/sub	Medium	Stream-scoped	Medium	Simple, existing Redis

What Works

• Design events, not messages: an event should describe what happened, not what the consumer should do. See Microservices Patterns for service communication strategies. OrderPlaced is correct. DecrementInventory is a command masquerading as an event. Events are facts; commands are instructions.
• Use schema validation: unvalidated events are a source of subtle bugs. Use Avro, JSON Schema, or Protobuf to define event contracts. Validate at the publisher and consumer boundaries. Version schemas and maintain backward compatibility — a Schema Registry does this centrally.
• Make consumers idempotent: network retries and broker redeliveries mean the same event can land more than once. See Idempotent Endpoints for deduplication patterns. Design handlers so that processing the same event twice produces the same state. Use UPSERT or track processed event IDs in a deduplication table.
• Monitor consumer lag: lag is the number of unprocessed messages in a partition. When lag keeps climbing, your consumers can’t keep up with the producers. Alert on lag thresholds, then scale consumers horizontally or optimize the handler.
• Publish domain events, not infrastructure events: PaymentProcessed is a domain event with business meaning. DatabaseRowInserted is infrastructure noise. Consumers care about business state changes, not implementation details.

Common mistakes

• Choreography without visibility: a request that fans out to 5 events, each triggering 3 more, creates an invisible workflow. When that chain breaks, good luck figuring out which of the 15 services started it. Add correlation IDs and distributed tracing to follow the chain.
• Synchronous event processing: a consumer that processes events synchronously inside an HTTP request reintroduces the coupling the event bus was meant to remove. Process events asynchronously, decoupled from the user-facing request.
• No error handling for poison messages: a malformed event that crashes the consumer gets redelivered forever and blocks the queue. Set a maximum retry count and route failures to a poison-pill handler.
• Storing state in the broker: using the broker as a database (querying Kafka for current state) is an anti-pattern. Brokers are for transport, not storage. Use event sourcing or a read model for state queries.

See Also

• Saga Pattern — compensating transactions when a workflow spans several services.
• CQRS Pattern — split read and write models once events become the source of truth.
• CloudEvents spec — a common envelope format if you want events that port across brokers.
• Runnable companion on GitHub — the companion repo for this recipe.