Event-Driven Microservices: Kafka, Saga Pattern, and Eventual Consistency

Distributed transactions are the hardest problem in microservices. The Saga pattern replaces one big ACID transaction with local transactions connected by events.

Choreography-Based Saga

from kafka import KafkaProducer, KafkaConsumer
import json, uuid

producer = KafkaProducer(
    bootstrap_servers=['kafka:9092'],
    value_serializer=lambda v: json.dumps(v).encode()
)

def create_order(customer_id: int, items: list, total: float) -> dict:
    order_id = str(uuid.uuid4())
    db.save_order({'order_id': order_id, 'status': 'pending'})
    producer.send('order-events', {
        'type': 'OrderCreated', 'order_id': order_id,
        'customer_id': customer_id, 'items': items, 'total': total
    })
    return {'order_id': order_id}

def process_events():
    consumer = KafkaConsumer(
        'inventory-events', 'payment-events',
        bootstrap_servers=['kafka:9092'],
        group_id='order-service',
        value_deserializer=lambda x: json.loads(x.decode())
    )
    for msg in consumer:
        event = msg.value
        if event['type'] == 'PaymentCharged':
            db.update_order(event['order_id'], status='confirmed')
        elif event['type'] == 'PaymentFailed':
            db.update_order(event['order_id'], status='failed')
            # Compensating: release inventory
            producer.send('inventory-events', {
                'type': 'ReleaseReservation', 'order_id': event['order_id']
            })

Idempotency

def handle_event(event: dict) -> None:
    event_id = event['event_id']
    if redis.sismember('processed_events', event_id):
        return  # Already processed
    process_order_event(event)
    redis.sadd('processed_events', event_id)
    redis.expire('processed_events', 86400)

Outbox Pattern (Guaranteed Delivery)

def create_order_safely(order_data: dict) -> None:
    with db.transaction():
        order = db.save_order(order_data)
        # Same transaction: both succeed or both fail
        db.save_outbox_event({
            'event_type': 'OrderCreated',
            'payload': json.dumps(order_data),
            'published': False
        })
    # Separate relay reads outbox and publishes to Kafka

Event Sourcing

class OrderEventStore:
    def append(self, order_id: str, event: dict) -> None:
        db.insert('order_events', {
            'order_id': order_id, 'event_type': event['type'],
            'event_data': json.dumps(event), 'created_at': datetime.utcnow()
        })

    def replay(self, order_id: str) -> dict:
        events = db.query(
            'SELECT * FROM order_events WHERE order_id = ? ORDER BY id',
            order_id
        )
        state = {}
        for e in events:
            ev = json.loads(e.event_data)
            if ev['type'] == 'OrderCreated':
                state = {**ev, 'status': 'pending'}
            elif ev['type'] == 'OrderConfirmed':
                state = {**state, 'status': 'confirmed'}
        return state

The Saga mindset: design workflows not transactions. Every step must be idempotent and have a compensating action.

→ Validate event schemas with the JSON Viewer tool.