Scenario

A production order system has to do more than move messages. Each order flows through several steps — reserve inventory, charge payment, ship — and any step can fail. The system must compensate for partial failures, isolate downstream outages, never double-charge a customer on a redelivered message, and reconstruct account state from history when something goes wrong.

KubeMQ exposes five messaging primitives — Events, Events Store, Queues, Commands, and Queries — and kubemq-faststream maps each to a @broker.subscriber keyword. The resilience patterns on this page are built entirely from those primitives plus queue parameters the broker already enforces (max_receive_count, max_receive_queue, delay_in_seconds, max_messages). No extra infrastructure is required.

This page assembles the production patterns from the examples/advanced_patterns/ and examples/patterns/ directories of the kubemq-faststream repository into one resilience playbook. Every snippet is runnable against a single broker.

kubemq-faststream talks to KubeMQ over the native gRPC port 50000 — the same transport the native SDKs use. There is no connector to enable and no HTTP flag to set. See Getting Started for the full setup.

Every example connects the same way and reads the broker address from KUBEMQ_ADDRESS, so you can point any of them at a remote broker without editing code:

common.py

import os
from faststream import FastStream
from kubemq_faststream import KubeMQBroker

KUBEMQ_ADDRESS = os.environ.get("KUBEMQ_ADDRESS", "kubemq://localhost:50000")

broker = KubeMQBroker(KUBEMQ_ADDRESS)
app = FastStream(broker)

Run a broker locally first:

docker run -d \  --name kubemq \  -p 50000:50000 \  -p 9090:9090 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  europe-docker.pkg.dev/kubemq/images/kubemq:next

Port 50000 is the gRPC port FastStream connects to. Port 9090 is the shared HTTP server (REST and connector endpoints) — not required here, but harmless to expose.

Saga Orchestration

A saga executes a sequence of steps and, if any step fails, runs compensating actions in reverse to undo the work already done. Here each step is a KubeMQ command (broker.request blocks until the handler replies), and each completed step records the channel that reverses it.

saga_pattern.py

@broker.subscriber(commands="example.saga.reserve_inventory")
async def reserve_inventory(msg: dict) -> dict:
    print("[saga] Step 1: reserve_inventory OK")
    return {"status": "reserved", "order_id": msg["order_id"]}


@broker.subscriber(commands="example.saga.charge_payment")
async def charge_payment(msg: dict) -> dict:
    print("[saga] Step 2: charge_payment OK")
    return {"status": "charged", "order_id": msg["order_id"]}


@broker.subscriber(commands="example.saga.ship_order")
async def ship_order(msg: dict) -> dict:
    print("[saga] Step 3: ship_order OK")
    return {"status": "shipped", "order_id": msg["order_id"]}


@broker.subscriber(commands="example.saga.compensate.inventory")
async def compensate_inventory(msg: dict) -> None:
    print(f"[compensate] Released inventory for order {msg['order_id']}")


@broker.subscriber(commands="example.saga.compensate.payment")
async def compensate_payment(msg: dict) -> None:
    print(f"[compensate] Refunded payment for order {msg['order_id']}")

The orchestrator walks the step list. Each step pairs a forward channel with its compensation channel (the final step needs none). On failure, it replays the recorded compensations in reverse order:

saga_pattern.py

SAGA_STEPS = [
    ("example.saga.reserve_inventory", "example.saga.compensate.inventory"),
    ("example.saga.charge_payment", "example.saga.compensate.payment"),
    ("example.saga.ship_order", None),  # last step has no compensation
]


async def run_saga(order_id: str) -> bool:
    """Execute saga steps in order; compensate on failure."""
    completed: list[str] = []

    for step_channel, compensate_channel in SAGA_STEPS:
        try:
            await broker.request(
                {"order_id": order_id},
                commands=step_channel,
                timeout=10,
            )
            if compensate_channel:
                completed.append(compensate_channel)
        except Exception as exc:
            print(f"[saga] Step failed ({step_channel}): {exc}")
            # Compensate in reverse order
            for comp_channel in reversed(completed):
                await broker.request(
                    {"order_id": order_id},
                    commands=comp_channel,
                    timeout=10,
                )
            return False

    return True

Commands are request-reply with a void result. The orchestrator's timeout=10 is per step — a step that never replies is treated as a failure and triggers compensation, so a hung downstream service cannot leave the saga half-finished.

Dead-Letter Processing

When a queue message keeps failing, you do not want it redelivered forever. KubeMQ enforces a per-message max_receive_count: after that many delivery attempts, the broker automatically routes the message to the max_receive_queue (the dead-letter queue). A separate subscriber drains the DLQ for investigation. The main handler uses AckPolicy.NACK_ON_ERROR so a raised exception nacks the message and counts as a failed attempt.

dead_letter_processing.py

from kubemq_faststream import AckPolicy, KubeMQBroker

MAIN_QUEUE = "example.advanced.dlq.main"
DLQ_QUEUE = "example.advanced.dlq"


@broker.subscriber(queues=MAIN_QUEUE, ack_policy=AckPolicy.NACK_ON_ERROR)
async def handle_order(msg: dict) -> None:
    """Intentionally fail to trigger DLQ routing after max receive count."""
    order_id = msg.get("order_id", "unknown")
    print(f"[main] Processing order: {order_id} -- simulated failure")
    raise ValueError(f"Cannot process order {order_id}")


@broker.subscriber(queues=DLQ_QUEUE)
async def handle_dead_letter(msg: dict) -> None:
    """Capture and log messages routed to the dead-letter queue."""
    print(f"[dlq] Dead-letter received: {msg}")
    print("[dlq] Logging failed order for manual review")

The DLQ routing lives on the publish call — max_receive_count and max_receive_queue are queue parameters, not subscriber settings:

dead_letter_processing.py

await broker.publish(
    {"order_id": "order-500", "item": "premium-widget", "amount": 99.99},
    queues=MAIN_QUEUE,
    max_receive_count=3,
    max_receive_queue=DLQ_QUEUE,
)

After three failed deliveries the broker moves the message to example.advanced.dlq, and handle_dead_letter logs it:

[main] Processing order: order-500 -- simulated failure
[main] Processing order: order-500 -- simulated failure
[main] Processing order: order-500 -- simulated failure
[dlq] Dead-letter received: {'order_id': 'order-500', ...}
[dlq] Logging failed order for manual review

For the full set of acknowledgement strategies and DLQ fundamentals, see the Queues page.

Retry with Exponential Backoff

Transient failures — a brief network blip, a rate-limited dependency — deserve a retry, not a dead letter. This pattern re-publishes the failed message with an increasing delay_in_seconds so retries are spaced out (1s, 2s, 4s, …). A max_receive_count ceiling still guarantees the message lands in the DLQ if it never succeeds.

retry_with_backoff.py

MAIN_QUEUE = "example.advanced.retry"
DLQ_QUEUE = "example.advanced.retry.dlq"

attempt_tracker: dict[str, int] = {}


@broker.subscriber(queues=MAIN_QUEUE)
async def handle_with_retry(msg: dict) -> None:
    """Process messages with simulated transient failures."""
    task_id = msg.get("task_id", "unknown")
    attempt_tracker[task_id] = attempt_tracker.get(task_id, 0) + 1
    attempt = attempt_tracker[task_id]

    if attempt < 3:
        print(f"[retry] Attempt for {task_id}, simulating transient failure")
        # Re-publish with exponential backoff delay
        delay = 2 ** (attempt - 1)  # 1s, 2s, 4s, ...
        await broker.publish(
            msg,
            queues=MAIN_QUEUE,
            max_receive_count=5,
            max_receive_queue=DLQ_QUEUE,
            delay_in_seconds=delay,
        )
        print(f"Published retry with delay={delay}s")
    else:
        print(f"[retry] Attempt for {task_id}, processing succeeded on retry")


@broker.subscriber(queues=DLQ_QUEUE)
async def handle_dead_letter(msg: dict) -> None:
    """Process messages that exceeded max receive count."""
    print(f"[dlq] Dead-letter received: {msg}")

The delay_in_seconds parameter uses the broker's delayed-delivery policy: the message is published immediately but only becomes visible to the subscriber after the delay elapses. Combining application-level retry counting with the broker's max_receive_count gives you both backoff control and a hard ceiling.

Circuit Breaker

A circuit breaker stops hammering a failing dependency. It tracks consecutive failures and, once a threshold is crossed, opens to reject work for a recovery window. After the window it goes half-open to test a single message: success closes the circuit, failure re-opens it. Here the breaker wraps a queue subscriber.

circuit_breaker.py

import enum
import time


class State(enum.Enum):
    CLOSED = "CLOSED"
    OPEN = "OPEN"
    HALF_OPEN = "HALF_OPEN"


class CircuitBreaker:
    """Simple circuit breaker with failure counting and timed recovery."""

    def __init__(self, failure_threshold: int = 3, recovery_timeout: float = 5.0) -> None:
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.state = State.CLOSED
        self.failure_count = 0
        self.last_failure_time: float = 0.0

    def allow_request(self) -> bool:
        if self.state == State.CLOSED:
            return True
        if self.state == State.OPEN:
            if time.monotonic() - self.last_failure_time >= self.recovery_timeout:
                self.state = State.HALF_OPEN
                print("[cb] Circuit HALF_OPEN -- testing recovery")
                return True
            return False
        # HALF_OPEN: allow one test request
        return True

    def record_success(self) -> None:
        self.failure_count = 0
        self.state = State.CLOSED

    def record_failure(self) -> None:
        self.failure_count += 1
        self.last_failure_time = time.monotonic()
        if self.failure_count >= self.failure_threshold:
            self.state = State.OPEN
            print("[cb] Circuit OPEN -- rejecting messages")

The subscriber consults the breaker before doing any work. While the circuit is open it returns early, leaving the dependency alone until the recovery timeout passes:

circuit_breaker.py

cb = CircuitBreaker(failure_threshold=3, recovery_timeout=3.0)


@broker.subscriber(queues="example.advanced.circuit")
async def handle(msg: dict) -> None:
    if not cb.allow_request():
        print(f"[cb] {cb.state.value}: rejecting msg")
        return

    try:
        do_work(msg)  # your real processing
        cb.record_success()
    except RuntimeError:
        cb.record_failure()

Idempotent Consumer

At-least-once delivery means a subscriber can see the same message twice — after a redelivery, a retry, or a duplicate publish. An idempotent consumer makes reprocessing a no-op by tracking the message_id of every message it has already handled. The message_id is set explicitly on publish so the value is stable across deliveries.

idempotent_consumer.py

CHANNEL = "example.advanced.idempotent"

# In-memory deduplication store (use Redis/DB in production)
processed_ids: set[str] = set()


@broker.subscriber(queues=CHANNEL)
async def idempotent_handler(msg: dict) -> None:
    """Process each unique message_id only once."""
    msg_id = msg.get("message_id", "")
    if not msg_id:
        print("[idempotent] No message_id in payload, processing anyway")
        return

    if msg_id in processed_ids:
        print(f"[idempotent] Duplicate skipped: {msg_id}")
        return

    processed_ids.add(msg_id)
    print(f"[idempotent] Processing new message: {msg_id}")

Publishing the same payload twice with an identical message_id exercises the guard:

idempotent_consumer.py

msg_id = "order-abc-001"
payload = {"message_id": msg_id, "order": "widget-x", "quantity": 5}

await broker.publish(payload, queues=CHANNEL, message_id=msg_id)
await broker.publish(payload, queues=CHANNEL, message_id=msg_id)  # duplicate

[idempotent] Processing new message: order-abc-001
[idempotent] Duplicate skipped: order-abc-001

The in-memory set resets when the process restarts. In production, back the dedup store with a durable, shared store (Redis, a database) and give entries a TTL so the set does not grow without bound.

Event Sourcing

Instead of storing current state, event sourcing stores the sequence of events that produced it, and rebuilds state by replaying them. KubeMQ's Events Store is the append-only log: publish domain events to an events_store channel, then subscribe with start_position=StartPosition.START_FROM_FIRST to replay the entire history from the beginning.

event_sourcing.py

from kubemq_faststream import KubeMQBroker, StartPosition

CHANNEL = "example.advanced.eventsourcing"

# Reconstructed state
account_state = {"balance": 0, "event_count": 0}


@broker.subscriber(
    events_store=CHANNEL,
    start_position=StartPosition.START_FROM_FIRST,
    group="event-sourcing-replay",
)
async def replay_events(msg: dict) -> None:
    """Replay events from the store to reconstruct account state."""
    event_type = msg.get("type", "unknown")
    amount = msg.get("amount", 0)
    account_state["event_count"] += 1
    count = account_state["event_count"]

    if event_type == "account_opened":
        account_state["balance"] = 0
    elif event_type == "deposit":
        account_state["balance"] += amount
    elif event_type == "withdrawal":
        account_state["balance"] -= amount

    print(f"[replay] Event {count}: {event_type} -- balance={account_state['balance']}")

Each domain event is appended to the store with a normal broker.publish:

event_sourcing.py

events = [
    {"type": "account_opened", "account_id": "acct-100"},
    {"type": "deposit", "account_id": "acct-100", "amount": 500},
    {"type": "withdrawal", "account_id": "acct-100", "amount": 200},
    {"type": "deposit", "account_id": "acct-100", "amount": 250},
]

for event in events:
    await broker.publish(event, events_store=CHANNEL)

Because the subscriber starts from START_FROM_FIRST, a fresh process replays the full ledger and arrives at the same balance every time:

[replay] Event 1: account_opened -- balance=0
[replay] Event 2: deposit -- balance=500
[replay] Event 3: withdrawal -- balance=300
[replay] Event 4: deposit -- balance=550

START_FROM_FIRST is one of several replay positions. To resume from a sequence number or a timestamp instead, see the Events Store page.

Distribution and Ordering

The remaining patterns control how work spreads across consumers and in what order it is processed.

Pipeline Composition

The foundational patterns in examples/patterns/ show how the primitives compose into multi-stage workflows. A pipeline chain moves a message through patterns by stage: an event triggers reliable queue processing, which fires a command for final execution.

pipeline_chain.py

STAGE_1 = "example.patterns.pipeline.ingest"
STAGE_2 = "example.patterns.pipeline.process"
STAGE_3 = "example.patterns.pipeline.execute"


@broker.subscriber(events=STAGE_1)
async def stage_ingest(msg: dict) -> None:
    """Stage 1: Receive event and forward to queue for reliable processing."""
    print(f"[Stage 1 - Event] Ingested: {msg}")
    enriched = {**msg, "ingested": True}
    await broker.publish(enriched, queues=STAGE_2)


@broker.subscriber(queues=STAGE_2)
async def stage_process(msg: dict) -> None:
    """Stage 2: Process from queue and trigger a command."""
    print(f"[Stage 2 - Queue] Processed: {msg}")
    await broker.request({**msg, "processed": True}, commands=STAGE_3, timeout=10)


@broker.subscriber(commands=STAGE_3)
async def stage_execute(msg: dict) -> None:
    """Stage 3: Execute the final command."""
    print(f"[Stage 3 - Command] Executed: {msg}")

To scale a single stage horizontally, register several subscribers on the same queue channel with the same group. KubeMQ delivers each message to exactly one member of the group — competing consumers load-balance the work:

competing_consumers.py

CHANNEL = "example.patterns.competing.tasks"


@broker.subscriber(queues=CHANNEL, group="workers")
async def worker_1(msg: dict) -> None:
    print(f"[worker-1] Processing task {msg.get('task_id', '?')}")


@broker.subscriber(queues=CHANNEL, group="workers")
async def worker_2(msg: dict) -> None:
    print(f"[worker-2] Processing task {msg.get('task_id', '?')}")


@broker.subscriber(queues=CHANNEL, group="workers")
async def worker_3(msg: dict) -> None:
    print(f"[worker-3] Processing task {msg.get('task_id', '?')}")

The examples/patterns/multi_pattern_app.py example wires all five primitives into one cohesive workflow — an event forwards to a queue, the queue fires a command, an Events Store subscriber persists the stream, and a query reports status — demonstrating that these patterns are not isolated demos but composable building blocks.

Running the Examples

Start a broker

docker run -d \  --name kubemq \  -p 50000:50000 \  -p 9090:9090 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  europe-docker.pkg.dev/kubemq/images/kubemq:next

Install the package

pip install kubemq-faststream

Run any pattern

Each file is self-contained and exits on its own after the demo completes:

python examples/advanced_patterns/saga_pattern.py
python examples/advanced_patterns/dead_letter_processing.py
python examples/advanced_patterns/circuit_breaker.py
python examples/advanced_patterns/event_sourcing.py

Point any example at a remote broker without touching code:

KUBEMQ_ADDRESS=kubemq://my-broker:50000 python examples/advanced_patterns/saga_pattern.py

Resilient Messaging Pipelines

Scenario

Saga Orchestration

Dead-Letter Processing

Retry with Exponential Backoff

Circuit Breaker

Idempotent Consumer

Event Sourcing

Distribution and Ordering

Pipeline Composition

Running the Examples

Start a broker

Install the package

Run any pattern

Queues

Events Store

Composition

On this page

Resilient Messaging Pipelines

Fan-out / fan-in with correlation tracking

Priority routing

Ordered processing

Cross-pattern correlation tracking

Queues

Events Store

Composition

On this page