How to Implement Event-Driven Architecture

This reference explains how to implement event-driven systems using standard messaging patterns, without relying on vendor-specific tooling or platform features. It covers the core mechanics of event publication, consumption, and processing in distributed applications.

Designing Event Contracts

Events must be defined with explicit schemas to ensure compatibility between producers and consumers. Every event should include:

A unique identifier (UUID)
A timestamp in ISO 8601 format
A version number for backward compatibility
A subject identifying the domain or entity
A payload containing the actual data

Event Schema Example

{
  "id": "123e4567-e89b-12d3-a456-426614174000",
  "timestamp": "2023-10-15T14:30:00Z",
  "version": "1.0",
  "subject": "user.created",
  "payload": {
    "userId": "user_12345",
    "email": "[email protected]",
    "createdAt": "2023-10-15T14:30:00Z"
  }
}

Event Validation Rules

Events must validate that the subject field follows the format domain.event-type where domain is a lowercase alphanumeric string and event-type is a lowercase hyphenated string. Events without proper formatting should be rejected with HTTP 400 status code.

Setting Up Message Brokers

Choose a message broker that supports durable queues and at-least-once delivery semantics. Common implementations include Apache Kafka, RabbitMQ, and AWS SQS.

Required Broker Configuration

For a system using Apache Kafka:

# server.properties
broker.id=0
listeners=PLAINTEXT://localhost:9092
advertised.listeners=PLAINTEXT://localhost:9092
log.dirs=/tmp/kafka-logs
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1

Topic Creation Command

# Create a topic with 3 partitions and 1 replica
bin/kafka-topics.sh --create \
  --topic user-events \
  --bootstrap-server localhost:9092 \
  --partitions 3 \
  --replication-factor 1 \
  --config retention.ms=604800000

Publishing Events

Event publishers must ensure that events are persisted to the message broker before acknowledging the request. Use synchronous publishing with confirmation mechanisms.

Publisher Implementation Example

import json
import uuid
from datetime import datetime
import pika

def publish_event(connection, exchange_name, event_data):
    channel = connection.channel()
    channel.exchange_declare(exchange=exchange_name, exchange_type='topic', durable=True)
    
    # Add metadata to the event
    event = {
        "id": str(uuid.uuid4()),
        "timestamp": datetime.utcnow().isoformat() + "Z",
        "version": "1.0",
        "subject": event_data["subject"],
        "payload": event_data["payload"]
    }
    
    # Publish with mandatory flag to catch failures
    channel.basic_publish(
        exchange=exchange_name,
        routing_key=event_data["subject"],
        body=json.dumps(event),
        properties=pika.BasicProperties(delivery_mode=2),  # Make message persistent
        mandatory=True
    )

Publisher Error Handling

If the broker rejects an event, the publisher should retry with exponential backoff or route the event to a dead-letter queue. The publisher must not acknowledge the request until confirmation is received.

Consuming Events

Consumers must implement idempotent processing to handle duplicate events. Each consumer should track processed event IDs to avoid reprocessing.

Consumer Configuration

# consumer-config.yaml
consumer:
  group_id: user-service-group
  auto_offset_reset: earliest
  enable_auto_commit: false
  max_poll_interval_ms: 300000

Event Processing Loop

import json
from confluent_kafka import Consumer, KafkaException

def process_events(consumer):
    try:
        while True:
            msg = consumer.poll(timeout=1.0)
            if msg is None:
                continue
            if msg.error():
                raise KafkaException(msg.error())
            
            event = json.loads(msg.value())
            event_id = event["id"]
            
            # Check if event was already processed
            if event_already_processed(event_id):
                consumer.commit()
                continue
                
            # Process the event
            try:
                process_event(event)
                mark_event_as_processed(event_id)
                consumer.commit()
            except Exception as e:
                # Log error and requeue
                log_error(event_id, str(e))
                raise  # Let consumer retry with backoff
                
    except KeyboardInterrupt:
        pass
    finally:
        consumer.close()

Error Handling Patterns

Dead Letter Queue Setup

Configure a dead letter exchange for failed events:

# Create dead letter exchange
bin/kafka-topics.sh --create \
  --topic user-events-dlq \
  --bootstrap-server localhost:9092 \
  --partitions 1 \
  --replication-factor 1

Retry Logic Implementation

Consumers should implement circuit breaker patterns for transient failures. When processing fails, events must be retried up to three times before moving to DLQ.

Event Routing and Filtering

Use topic-based routing with wildcards where supported. Implement filters on the consumer side to handle only relevant events.

Topic Subscription Example

# Subscribe to multiple event types
consumer.subscribe([
    "user.*",           # All user-related events
    "order.created",    # Specific order event
    "payment.*"         # All payment events
])

Backward Compatibility

Version all events with a version field. When changing event structure, introduce new versions rather than modifying existing ones.

Version Migration Strategy

When upgrading consumers:

Deploy new version alongside old version
Send both old and new version events for a period
Gradually shift traffic to new version
Decommission old version after transition period

Monitoring and Observability

Required Metrics

Event throughput (events/sec)
Processing latency (ms)
Consumer lag (number of unprocessed messages)
Error rates by event type

Health Check Endpoint

@app.get("/health")
def health_check():
    # Verify connection to message broker
    try:
        producer.flush(timeout=1.0)
        return {"status": "healthy", "timestamp": datetime.utcnow().isoformat()}
    except Exception as e:
        return {"status": "unhealthy", "error": str(e)}, 500

Common Failure Modes

Silent Data Loss

Events can be lost if producers don't wait for confirmation. If the broker receives a message but the producer doesn't receive acknowledgment, the event is effectively lost.

Duplicate Processing

Consumers must implement idempotent processing. Without proper deduplication logic, processing the same event twice can cause data inconsistency.

Consumer Group Rebalancing

When consumers join or leave a group, partitions are reassigned. During this period, some events may be processed multiple times or not at all.

Dead Letter Queue Issues

Events in DLQ must be manually reviewed and reprocessed. Without proper monitoring, failed events can accumulate without detection.

Performance Considerations

Batch Processing

For high-volume systems, implement batch processing to reduce overhead:

def process_batch(events):
    # Process multiple events in single transaction
    for event in events:
        process_event(event)
    
    # Commit all at once
    commit_batch()

Memory Management

Consumers must not hold references to processed events indefinitely. Implement proper cleanup after successful processing.

Deployment Strategy

Rolling Updates

Deploy consumers in rolling fashion to avoid service interruption. Ensure that both old and new versions can process the same event types.

Zero Downtime Migration

When migrating from one event format to another, implement parallel processing:

Both old and new event handlers run
New handler processes events with new format
Old handler continues handling old format
Gradually phase out old handler

Testing Event-Driven Systems

Integration Tests

Test complete event flows from producer through broker to consumer:

def test_event_flow():
    # Publish an event
    event = {"subject": "user.created", "payload": {"userId": "test123"}}
    publish_event(connection, "events", event)
    
    # Verify it was processed
    assert event_processed("test123")

End-to-End Test Scenarios

Event published → consumed successfully
Consumer failure → event reprocessed after restart
Broker failure → events persisted and replayed
Schema change → backward compatibility maintained

Configuration Management

Environment-Specific Settings

# config.yaml
event_broker:
  host: "localhost"
  port: 9092
  topics:
    user_events: "user-events"
    order_events: "order-events"

Configuration Validation

All event-related configuration must be validated at startup. Missing or invalid topic names should cause application startup to fail.

Security Considerations

Event Signing

Implement event signing with HMAC-SHA256 for integrity verification:

import hmac
import hashlib

def sign_event(event, secret_key):
    message = json.dumps(event, sort_keys=True).encode('utf-8')
    signature = hmac.new(
        secret_key.encode('utf-8'),
        message,
        hashlib.sha256
    ).hexdigest()
    return signature

Authentication

Consumers must authenticate with the message broker using appropriate credentials. Event producers should validate that they have permission to publish to specific topics.

Scaling Considerations

Partition Strategy

Ensure sufficient partitions for throughput. Each partition can be consumed by only one consumer in a group, so increase partitions to enable parallel processing.

Consumer Scaling

Scale consumers horizontally by increasing the number of instances in a consumer group. Each instance should process a subset of partitions.

Resource Management

Monitor broker memory usage and adjust partition counts accordingly. Excessive partitions can cause performance degradation.

This page was rewritten on 15 October 2026. It replaced a templated version that was not specific to its own title. How these pages are written and checked. If anything here is wrong, tell us at [email protected] and we will correct it.