Message Queue Best Practices for Production Systems

Message queues help distributed systems process work asynchronously, absorb traffic spikes, isolate failures, and decouple producers from consumers. They are commonly used for background jobs, order processing, notifications, logistics workflows, data pipelines, payments, and communication between microservices.

Adding a queue does not automatically make a system reliable. Production messaging requires clear delivery guarantees, idempotent consumers, bounded retries, dead-letter handling, controlled concurrency, useful observability, and operational procedures for replaying failed messages. Without these controls, queues can hide failures, multiply duplicate side effects, and turn temporary outages into large backlogs.

Table of Contents

• Design for Failure
• Choose the Right Delivery Guarantee
  • At-Most-Once Delivery
  • At-Least-Once Delivery
  • Exactly-Once Claims
• Make Consumers Idempotent
• Use Stable Message Identifiers
• Design Clear Message Contracts
  • Message Envelope
  • Commands vs Events
  • Schema Versioning
• Keep Message Payloads Small
• Validate Messages at the Boundary
• Configure Retries Carefully
  • Classify Failures
  • Use Exponential Backoff
  • Add Jitter
  • Set Retry Limits
• Use Dead-Letter Queues
• Handle Poison Messages
• Control Consumer Concurrency
• Apply Backpressure
• Manage Message Visibility and Leases
• Acknowledge Messages After Success
• Preserve Ordering Only Where Required
• Partition Messages Intentionally
• Avoid Hot Partitions
• Design for Replay
• Use the Transactional Outbox Pattern
• Protect Downstream Dependencies
• Monitor Queue Health
• Use Structured Logging and Tracing
• Scale Consumers Safely
• Plan for Backlog Recovery
• Set Retention and Expiration Rules
• Secure the Messaging System
• Test Real Failure Scenarios
• Real-World Order Processing Example
• Common Mistakes
• Production Checklist
• Conclusion
• Key Takeaway
• More Articles to Read

Design for Failure

A production message queue should be treated as part of a distributed failure-handling system, not only as a transport layer. Messages may be delivered more than once, delayed, processed out of order, rejected, or left unacknowledged after a consumer crashes.

Failures can occur at every stage:

• The producer may fail before publishing.
• The producer may time out even though the broker accepted the message.
• The broker may temporarily reject writes.
• The consumer may crash during processing.
• A downstream database or API may be unavailable.
• An acknowledgement may fail after business work succeeds.
• A malformed message may fail every time it is retried.

A reliable design begins by assuming that these situations will happen. The system should preserve important work, avoid uncontrolled retries, prevent duplicate business effects, and make failures visible to operators.

Choose the Right Delivery Guarantee

Delivery guarantees describe how the broker and consumer handle message loss and duplication. No single model is best for every workload.

At-Most-Once Delivery

A message is delivered zero or one time. The system does not retry after uncertain failures.

Advantages

• Low processing overhead.
• No broker-generated duplicate delivery.
• Useful for disposable or frequently refreshed data.

Disadvantages

• Messages can be lost.
• Temporary failures become permanent data gaps.
• Not suitable for critical business operations.

When to Use

Use at-most-once delivery when losing an occasional message is acceptable and the next update replaces the missing one.

Real-World Use Cases

• Live cursor positions.
• Temporary presence updates.
• High-frequency telemetry samples.

Example

def handle_presence_update(message: dict) -> None:
    # A later update replaces the current state.
    presence_cache.set(
        key=message["user_id"],
        value=message["status"],
        ttl_seconds=60
    )

At-Least-Once Delivery

A message is retried until it is acknowledged or reaches a failure policy. The same message may be delivered multiple times.

Advantages

• Temporary failures do not automatically lose work.
• Suitable for important business operations.
• Supported by most queue systems.

Disadvantages

• Consumers must handle duplicates.
• Incorrect acknowledgement logic can repeat side effects.
• Retries can overload failing dependencies.

When to Use

Use at-least-once delivery for orders, payments, notifications, logistics operations, data processing, and most production background jobs.

Real-World Use Cases

• Creating shipment records.
• Sending customer invoices.
• Updating search indexes.
• Processing uploaded documents.

Example

def process_message(message: dict) -> None:
    try:
        handle_business_operation(message)
        queue.acknowledge(message["receipt_handle"])
    except Exception:
        # The message remains unacknowledged and may be delivered again.
        raise

Exactly-Once Claims

Exactly-once delivery is often misunderstood. A broker may provide deduplicated publication or transactional consumption inside its own boundaries, but complete business processing usually includes databases, APIs, emails, payment providers, and other systems.

If a consumer updates a database and crashes before acknowledging the message, the broker may redeliver it. The database update already happened, so exactly-once business behaviour still requires idempotency.

Advantages

• Can reduce duplicate handling inside a supported platform.
• Useful for stream-processing pipelines with transactional state.

Disadvantages

• Usually limited to specific broker operations.
• Does not automatically cover external side effects.
• Adds operational and conceptual complexity.

When to Use

Use broker-level exactly-once features when the workload remains within the platform’s transactional model. Continue using idempotency for external systems.

Real-World Use Cases

• Transactional stream processing.
• Broker-to-broker transformations.
• Materialised views managed inside one streaming platform.

Make Consumers Idempotent

An idempotent consumer can process the same message multiple times without creating multiple business effects.

Consider a consumer that creates an invoice. If the message is delivered twice and each attempt inserts a new invoice, the customer may receive duplicate charges or documents.

A processed-message table is one common solution.

CREATE TABLE processed_messages (
    consumer_name VARCHAR(150) NOT NULL,
    message_id UUID NOT NULL,
    processed_at TIMESTAMP NOT NULL DEFAULT NOW(),
    PRIMARY KEY (consumer_name, message_id)
);

The deduplication insert and business update should occur in the same transaction.

def handle_order_confirmed(database, message: dict) -> None:
    with database.transaction() as transaction:
        inserted = transaction.execute(
            """
            INSERT INTO processed_messages (
                consumer_name,
                message_id
            )
            VALUES (%s, %s)
            ON CONFLICT DO NOTHING
            """,
            [
                "invoice-order-consumer",
                message["message_id"]
            ]
        )

        if inserted.row_count == 0:
            return

        transaction.execute(
            """
            INSERT INTO invoices (
                order_id,
                customer_id,
                status
            )
            VALUES (%s, %s, 'pending')
            ON CONFLICT (order_id) DO NOTHING
            """,
            [
                message["order_id"],
                message["customer_id"]
            ]
        )

Business-level unique constraints provide another layer of protection. A unique order ID on the invoice table prevents duplicate invoice creation even if deduplication metadata is unavailable.

Use Stable Message Identifiers

Every message should have a globally unique identifier generated once by the producer. Retries, redelivery, replay, and dead-letter transfers should preserve that identifier.

import uuid
from datetime import datetime, timezone


def create_message(event_type: str, data: dict) -> dict:
    return {
        "message_id": str(uuid.uuid4()),
        "event_type": event_type,
        "event_version": 1,
        "occurred_at": datetime.now(timezone.utc).isoformat(),
        "data": data
    }

Generating a new ID during every retry makes duplicate detection impossible because consumers see each attempt as a different business event.

Separate identifiers may be useful for different concepts:

• message_id: identifies one logical message.
• correlation_id: connects related operations across services.
• causation_id: identifies the message or request that caused this message.
• aggregate_id: identifies the related business entity.

Design Clear Message Contracts

A message contract should describe what happened, which entity is affected, when the event occurred, and how consumers should interpret the payload.

Message Envelope

A consistent envelope simplifies routing, validation, logging, tracing, and schema evolution.

message = {
    "message_id": "455be1b8-e965-4694-ae62-e03832f85572",
    "event_type": "shipment.created",
    "event_version": 1,
    "occurred_at": "2026-07-27T15:30:00Z",
    "producer": "shipment-service",
    "correlation_id": "order-8192",
    "causation_id": "command-7741",
    "aggregate_type": "shipment",
    "aggregate_id": "shipment-4412",
    "data": {
        "order_id": "order-8192",
        "carrier_code": "carrier-a",
        "service_level": "express"
    }
}

Metadata should remain consistent across message types. Business fields belong inside the data section.

Commands vs Events

Commands request that a specific action be performed. Events describe something that already happened.

Avoid ambiguous names such as process.order unless the message is intentionally a command. Event names should normally use past-tense business language.

Schema Versioning

Messages may remain in queues during outages and can be replayed months later. Consumers should not assume every message follows the newest schema.

Safe evolution practices include:

• Add optional fields instead of changing existing meanings.
• Preserve old field names during migration.
• Include an explicit event version.
• Support old and new versions during rolling deployments.
• Create a new event type when semantics change significantly.

def parse_order_event(message: dict) -> dict:
    version = message.get("event_version", 1)

    if version == 1:
        return {
            "order_id": message["data"]["order_id"],
            "amount": message["data"]["amount"],
            "currency": message["data"]["currency"]
        }

    if version == 2:
        return {
            "order_id": message["data"]["order_id"],
            "amount": message["data"]["total"]["amount"],
            "currency": message["data"]["total"]["currency"]
        }

    raise ValueError(f"Unsupported event version: {version}")

Keep Message Payloads Small

Large messages increase network traffic, broker storage, processing latency, retry cost, and memory use. They may also exceed broker limits.

Messages should contain enough data for consumers to act correctly without turning the broker into file storage.

For large objects, store the content in object storage and publish a reference.

message = {
    "message_id": generate_uuid(),
    "event_type": "document.uploaded",
    "event_version": 1,
    "data": {
        "document_id": "document-771",
        "storage_key": "documents/2026/07/document-771.pdf",
        "content_type": "application/pdf",
        "size_bytes": 8421937
    }
}

The referenced object must remain available long enough for retries and replay. Access permissions should be based on the consumer’s identity rather than temporary credentials embedded in the message.

Validate Messages at the Boundary

Consumers should validate the envelope and payload before performing business work. Invalid messages should not reach database updates or external APIs.

from datetime import datetime
from typing import Any

from pydantic import BaseModel, Field


class ShipmentCreatedData(BaseModel):
    shipment_id: str = Field(min_length=1)
    order_id: str = Field(min_length=1)
    carrier_code: str = Field(min_length=1)


class ShipmentCreatedMessage(BaseModel):
    message_id: str
    event_type: str
    event_version: int
    occurred_at: datetime
    data: ShipmentCreatedData


def validate_message(payload: dict[str, Any]) -> ShipmentCreatedMessage:
    message = ShipmentCreatedMessage.model_validate(payload)

    if message.event_type != "shipment.created":
        raise ValueError("Unexpected event type")

    if message.event_version != 1:
        raise ValueError("Unsupported event version")

    return message

Validation failures are usually permanent. Retrying a payload with a missing required field will not repair it. Such messages should move to a dead-letter path or failed-message store.

Configure Retries Carefully

Retries are necessary for transient failures, but uncontrolled retries can make an outage worse.

Classify Failures

Failures should be separated into transient, permanent, and unknown categories.

class TransientProcessingError(Exception):
    pass


class PermanentProcessingError(Exception):
    pass


def process_carrier_request(message: dict) -> None:
    response = carrier_client.create_shipment(message["data"])

    if response.status_code in {429, 500, 502, 503, 504}:
        raise TransientProcessingError(
            f"Carrier temporarily unavailable: {response.status_code}"
        )

    if response.status_code in {400, 404, 422}:
        raise PermanentProcessingError(
            f"Carrier rejected request: {response.status_code}"
        )

    response.raise_for_status()

Use Exponential Backoff

Exponential backoff increases the delay after each failed attempt.

def calculate_retry_delay(
    attempt: int,
    base_seconds: int = 5,
    maximum_seconds: int = 900
) -> int:
    delay = base_seconds * (2 ** max(0, attempt - 1))
    return min(delay, maximum_seconds)

Example delays with a five-second base are 5, 10, 20, 40, 80, and 160 seconds.

This gives the failing dependency time to recover and reduces repeated load.

Add Jitter

If thousands of messages fail at the same time, identical retry schedules cause them to retry together. Jitter adds randomness so recovery traffic is distributed.

import random


def calculate_retry_delay_with_jitter(
    attempt: int,
    base_seconds: int = 5,
    maximum_seconds: int = 900
) -> float:
    exponential_delay = min(
        base_seconds * (2 ** max(0, attempt - 1)),
        maximum_seconds
    )

    return random.uniform(
        exponential_delay * 0.75,
        exponential_delay * 1.25
    )

Set Retry Limits

Every retry policy should define:

• Maximum attempt count.
• Maximum message age.
• Minimum and maximum delay.
• Retryable exception types.
• Final failure destination.

A message should not retry forever. Permanent failures consume capacity, increase costs, and can block ordered workloads.

Use Dead-Letter Queues

A dead-letter queue stores messages that cannot be processed successfully after the retry policy is exhausted.

Dead-letter records should preserve:

• Original message payload.
• Original message ID.
• Source queue or topic.
• Consumer name.
• Attempt count.
• First and last failure times.
• Error category and summary.
• Correlation and aggregate identifiers.

def create_dead_letter_record(
    message: dict,
    source_queue: str,
    consumer_name: str,
    attempt_count: int,
    error: Exception
) -> dict:
    return {
        "original_message": message,
        "source_queue": source_queue,
        "consumer_name": consumer_name,
        "attempt_count": attempt_count,
        "failure_type": error.__class__.__name__,
        "failure_message": str(error)[:2000],
        "message_id": message["message_id"],
        "correlation_id": message.get("correlation_id")
    }

A dead-letter queue is not a deletion mechanism. Failed messages require monitoring, ownership, investigation, and controlled replay procedures.

Handle Poison Messages

A poison message fails every time because the payload or business state makes successful processing impossible.

Common causes include:

• Malformed JSON.
• Unsupported schema version.
• Missing required identifiers.
• Invalid business transitions.
• Payloads exceeding downstream limits.
• Consumer bugs triggered by specific data.

Poison messages should be isolated quickly. Repeatedly returning them to the active queue wastes resources and may block other work.

def handle_queue_message(message: dict) -> None:
    try:
        validated_message = validate_message(message)
        process_valid_message(validated_message)

    except PermanentProcessingError as error:
        dead_letter_queue.publish(
            create_dead_letter_record(
                message=message,
                source_queue="shipment-events",
                consumer_name="shipment-worker",
                attempt_count=message.get("attempt_count", 1),
                error=error
            )
        )

        queue.acknowledge(message["receipt_handle"])

    except TransientProcessingError:
        raise

Control Consumer Concurrency

Increasing the number of consumers improves throughput only while downstream systems can handle the additional load.

Consumer concurrency should consider:

• Database connection limits.
• External API quotas.
• CPU and memory per message.
• Message-processing duration.
• Ordering requirements.
• Lock contention.

A semaphore can limit concurrent calls to a fragile dependency.

import asyncio


carrier_api_limit = asyncio.Semaphore(20)


async def process_shipment_message(message: dict) -> None:
    async with carrier_api_limit:
        await carrier_client.create_shipment(
            message["data"]
        )

Concurrency limits should be configurable so operators can reduce pressure during incidents without redeploying code.

Apply Backpressure

Backpressure prevents consumers from accepting more work than the application or its dependencies can process safely.

Useful mechanisms include:

• Limiting prefetched messages.
• Reducing consumer concurrency.
• Pausing queue consumption.
• Using rate limiters for downstream calls.
• Rejecting low-priority work during overload.
• Separating workloads into different queues.

Without backpressure, one consumer can reserve thousands of messages, hold them in memory, and delay redelivery if it crashes.

MAX_IN_FLIGHT_MESSAGES = 100


def start_consumer() -> None:
    queue.consume(
        queue_name="order-processing",
        handler=handle_order,
        prefetch_count=MAX_IN_FLIGHT_MESSAGES
    )

Manage Message Visibility and Leases

Many queue systems hide a message temporarily after a consumer receives it. If the consumer does not acknowledge before the visibility timeout expires, the message becomes available again.

The timeout must exceed normal processing time. If processing takes two minutes but visibility lasts thirty seconds, the same message may be handled concurrently by multiple consumers.

For variable-duration jobs, the consumer can extend the lease periodically.

import threading
import time


def extend_visibility_until_complete(
    queue,
    receipt_handle: str,
    stop_event: threading.Event
) -> None:
    while not stop_event.wait(timeout=30):
        queue.extend_visibility(
            receipt_handle=receipt_handle,
            timeout_seconds=60
        )


def process_long_running_message(message: dict) -> None:
    stop_event = threading.Event()

    heartbeat = threading.Thread(
        target=extend_visibility_until_complete,
        args=(
            queue,
            message["receipt_handle"],
            stop_event
        ),
        daemon=True
    )

    heartbeat.start()

    try:
        run_document_processing(message["data"])
        queue.acknowledge(message["receipt_handle"])
    finally:
        stop_event.set()
        heartbeat.join()

Lease extension should stop if the worker loses ownership or cannot contact the broker.

Acknowledge Messages After Success

A message should normally be acknowledged only after all required business work is committed.

def consume_payment_event(message: dict) -> None:
    process_payment_event(message)
    queue.acknowledge(message["receipt_handle"])

Acknowledging before processing risks message loss if the consumer crashes afterward.

For workflows with multiple steps, define what counts as successful processing. Optional analytics or logging should not prevent acknowledgement of a completed critical business transaction.

Preserve Ordering Only Where Required

Ordering reduces parallelism. A queue that guarantees global order may allow only one effective processing sequence, limiting throughput and availability.

Most systems need ordering only for related messages.

For example, events for one order may require this sequence:

1. order.created
2. order.confirmed
3. order.shipped

Events for different orders can be processed concurrently.

Use an aggregate ID or partition key to preserve local ordering.

queue.publish(
    topic="order-events",
    partition_key=event["aggregate_id"],
    message=event
)

Consumers can also use aggregate sequence numbers to reject stale events.

UPDATE order_projections
SET
    status = :status,
    aggregate_version = :incoming_version
WHERE order_id = :order_id
  AND aggregate_version < :incoming_version;

Partition Messages Intentionally

Partitioning spreads queue traffic across broker resources and consumer instances. The partition key determines which messages share ordering and load.

Common keys include:

• Order ID.
• Customer ID.
• Shipment ID.
• Tenant ID.
• Account ID.

The key should align with business ordering requirements and produce reasonably balanced traffic.

Avoid Hot Partitions

A hot partition receives much more traffic than others. One large tenant, customer, or aggregate can limit total throughput even when other partitions are idle.

Mitigation strategies include:

• Increase partition count.
• Use a more distributed partition key.
• Separate very large tenants.
• Hash keys when strict ordering is unnecessary.
• Split high-volume workloads by category or region.

import hashlib


def create_partition_key(
    tenant_id: str,
    order_id: str,
    partition_count: int
) -> str:
    raw_key = f"{tenant_id}:{order_id}"
    digest = hashlib.sha256(raw_key.encode()).hexdigest()
    partition_number = int(digest[:8], 16) % partition_count

    return f"partition-{partition_number}"

Hashing improves distribution but removes natural per-tenant ordering unless the key still includes the required aggregate boundary.

Design for Replay

Replay sends previously failed or historical messages through a consumer again. It is useful after bug fixes, dependency recovery, or rebuilding projections.

Safe replay should preserve:

• Original message ID.
• Original business timestamp.
• Original schema version.
• Correlation metadata.
• Replay reason and operator identity.

Replay should not silently create new logical events.

def replay_message(
    original_message: dict,
    reason: str,
    requested_by: str
) -> None:
    replay_headers = {
        "is_replay": "true",
        "replay_reason": reason,
        "replay_requested_by": requested_by
    }

    queue.publish(
        queue_name="order-events-replay",
        message=original_message,
        headers=replay_headers
    )

Bulk replay should use rate limits and a separate queue so historical traffic does not overwhelm live production work.

Use the Transactional Outbox Pattern

A producer that updates a database and publishes a message faces a dual-write problem. The database update may commit while message publication fails.

The transactional outbox pattern stores the business update and outgoing message in the same database transaction.

def confirm_order(database, order_id: str) -> None:
    with database.transaction() as transaction:
        transaction.execute(
            """
            UPDATE orders
            SET status = 'confirmed'
            WHERE order_id = %s
            """,
            [order_id]
        )

        transaction.execute(
            """
            INSERT INTO outbox_messages (
                outbox_id,
                event_type,
                aggregate_id,
                payload_json,
                status,
                created_at
            )
            VALUES (
                gen_random_uuid(),
                'order.confirmed',
                %s,
                %s,
                'pending',
                NOW()
            )
            """,
            [
                order_id,
                serialize_json({
                    "order_id": order_id
                })
            ]
        )

A separate publisher reads pending records and sends them to the broker. This prevents committed business changes from silently losing their events.

Protect Downstream Dependencies

Queues can absorb traffic faster than downstream services can process it. A sudden backlog replay may overload a database or third-party API.

Protection mechanisms include:

• Per-dependency concurrency limits.
• Token-bucket rate limiting.
• Circuit breakers.
• Timeouts.
• Bulkheads.
• Fallback queues.

import time


class CircuitBreaker:
    def __init__(
        self,
        failure_limit: int,
        recovery_seconds: int
    ) -> None:
        self.failure_limit = failure_limit
        self.recovery_seconds = recovery_seconds
        self.failure_count = 0
        self.opened_at: float | None = None

    def allow_request(self) -> bool:
        if self.opened_at is None:
            return True

        if time.time() - self.opened_at >= self.recovery_seconds:
            self.failure_count = 0
            self.opened_at = None
            return True

        return False

    def record_failure(self) -> None:
        self.failure_count += 1

        if self.failure_count >= self.failure_limit:
            self.opened_at = time.time()

    def record_success(self) -> None:
        self.failure_count = 0
        self.opened_at = None

When a circuit is open, messages should be delayed rather than retried immediately.

Monitor Queue Health

Consumer uptime does not prove that messages are being processed successfully. Monitoring should focus on progress and business latency.

The age of the oldest message is often more useful than total queue depth. A small queue can still contain one critical message stuck for hours.

Use Structured Logging and Tracing

Every processing attempt should produce structured logs with consistent fields.

import logging


logger = logging.getLogger(__name__)


def log_processing_failure(
    message: dict,
    attempt: int,
    error: Exception
) -> None:
    logger.exception(
        "Queue message processing failed",
        extra={
            "message_id": message.get("message_id"),
            "event_type": message.get("event_type"),
            "aggregate_id": message.get("aggregate_id"),
            "correlation_id": message.get("correlation_id"),
            "attempt": attempt,
            "error_type": error.__class__.__name__
        }
    )

Useful fields include:

• Message ID.
• Event or command type.
• Queue name.
• Consumer name.
• Attempt number.
• Aggregate ID.
• Correlation ID.
• Processing duration.
• Outcome.
• Error classification.

Distributed traces should continue across producers and consumers by passing tracing headers in the message envelope.

Scale Consumers Safely

Consumer autoscaling should use queue-specific signals rather than CPU alone.

Useful scaling inputs include:

• Queue depth per consumer.
• Oldest-message age.
• Arrival rate.
• Average processing duration.
• Downstream dependency capacity.

A simplified capacity estimate is:

import math


def required_consumers(
    arrival_rate_per_second: float,
    average_processing_seconds: float,
    concurrency_per_consumer: int,
    utilisation_target: float = 0.7
) -> int:
    capacity_per_consumer = (
        concurrency_per_consumer
        / average_processing_seconds
        * utilisation_target
    )

    return max(
        1,
        math.ceil(
            arrival_rate_per_second
            / capacity_per_consumer
        )
    )

Scaling must respect database connection limits and external API quotas. Increasing workers beyond dependency capacity creates more retries instead of more throughput.

Plan for Backlog Recovery

A broker or downstream outage can create millions of waiting messages. When the dependency recovers, processing the backlog at maximum speed may cause another outage.

A recovery plan should define:

• Maximum replay rate.
• Live-traffic priority.
• Consumer concurrency during recovery.
• External API quota allocation.
• Maximum acceptable message age.
• Rules for expiring obsolete work.

Separate live and recovery queues can protect current traffic.

def choose_processing_queue(message_age_seconds: int) -> str:
    if message_age_seconds > 3600:
        return "order-events-backlog"

    return "order-events-live"

Set Retention and Expiration Rules

Not every message remains useful forever. A delayed password-reset email, temporary cache refresh, or live-status update may become obsolete.

Each queue should define:

• Message retention period.
• Business expiration time.
• Dead-letter retention period.
• Replay eligibility window.

from datetime import datetime, timezone


def is_message_expired(message: dict) -> bool:
    expires_at = message.get("expires_at")

    if expires_at is None:
        return False

    expiration = datetime.fromisoformat(
        expires_at.replace("Z", "+00:00")
    )

    return datetime.now(timezone.utc) > expiration

Expired messages should be recorded as expired rather than reported as successful business processing.

Secure the Messaging System

Message brokers often contain operational, customer, financial, or logistics data. Security controls should apply to producers, consumers, administrative tools, and replay workflows.

Production controls include:

• Encryption in transit.
• Encryption at rest.
• Least-privilege publish and consume permissions.
• Separate credentials per application.
• Restricted dead-letter access.
• Audit logging for replay and deletion.
• Payload minimisation.
• Secret exclusion.
• Network access controls.

Consumers should not trust a message only because it came from the broker. Permissions, validation, tenant boundaries, and business authorisation still matter.

Test Real Failure Scenarios

Queue systems should be tested under the failures they are intended to survive.

Important tests include:

• Consumer crash before processing.
• Consumer crash after database commit.
• Acknowledgement timeout.
• Broker outage.
• Database outage.
• External API rate limiting.
• Duplicate delivery.
• Out-of-order delivery.
• Poison message handling.
• Dead-letter replay.
• Visibility timeout expiration.
• Large backlog recovery.
• Schema-version compatibility.
• Consumer autoscaling.

def test_duplicate_message_creates_one_invoice(
    database,
    invoice_consumer
):
    message = {
        "message_id": "73a3c72e-68f2-4dda-90e4-f15ba0fba39d",
        "event_type": "order.confirmed",
        "event_version": 1,
        "order_id": "order-101",
        "customer_id": "customer-81"
    }

    invoice_consumer.handle(message)
    invoice_consumer.handle(message)

    invoice_count = database.fetch_value(
        """
        SELECT COUNT(*)
        FROM invoices
        WHERE order_id = 'order-101'
        """
    )

    assert invoice_count == 1

Real-World Order Processing Example

Consider an order platform that publishes an order.confirmed event. Separate consumers handle inventory reservation, invoice creation, shipment planning, notifications, and analytics.

The producer stores the order update and outbox record atomically.

BEGIN;

UPDATE orders
SET
    status = 'confirmed',
    confirmed_at = NOW()
WHERE order_id = :order_id
  AND status = 'pending';

INSERT INTO outbox_messages (
    outbox_id,
    event_type,
    aggregate_id,
    payload_json,
    status,
    created_at
)
VALUES (
    gen_random_uuid(),
    'order.confirmed',
    :order_id,
    jsonb_build_object(
        'order_id', :order_id,
        'customer_id', :customer_id,
        'total_amount', :total_amount,
        'currency', :currency
    ),
    'pending',
    NOW()
);

COMMIT;

The inventory consumer validates and deduplicates the message.

def reserve_inventory(database, message: dict) -> None:
    validated = validate_order_confirmed(message)

    with database.transaction() as transaction:
        inserted = transaction.execute(
            """
            INSERT INTO processed_messages (
                consumer_name,
                message_id
            )
            VALUES (
                'inventory-order-consumer',
                %s
            )
            ON CONFLICT DO NOTHING
            """,
            [validated.message_id]
        )

        if inserted.row_count == 0:
            return

        for item in validated.data.items:
            available_quantity = transaction.fetch_value(
                """
                SELECT available_quantity
                FROM inventory
                WHERE product_id = %s
                FOR UPDATE
                """,
                [item.product_id]
            )

            if available_quantity < item.quantity:
                raise PermanentProcessingError(
                    f"Insufficient inventory for {item.product_id}"
                )

            transaction.execute(
                """
                UPDATE inventory
                SET available_quantity = available_quantity - %s
                WHERE product_id = %s
                """,
                [
                    item.quantity,
                    item.product_id
                ]
            )

        transaction.execute(
            """
            INSERT INTO inventory_reservations (
                order_id,
                status,
                created_at
            )
            VALUES (%s, 'reserved', NOW())
            ON CONFLICT (order_id) DO NOTHING
            """,
            [validated.data.order_id]
        )

The shipment consumer calls an external carrier API with a stable idempotency key.

def create_carrier_shipment(message: dict) -> None:
    idempotency_key = (
        f"create-shipment:{message['data']['order_id']}"
    )

    response = carrier_client.create_shipment(
        payload={
            "order_id": message["data"]["order_id"],
            "destination": message["data"]["destination"]
        },
        idempotency_key=idempotency_key,
        timeout_seconds=10
    )

    if response.status_code == 429:
        raise TransientProcessingError(
            "Carrier rate limit exceeded"
        )

    if response.status_code >= 500:
        raise TransientProcessingError(
            "Carrier service unavailable"
        )

    if response.status_code >= 400:
        raise PermanentProcessingError(
            "Carrier rejected shipment request"
        )

If the carrier times out after creating the shipment, the stable idempotency key prevents the retry from creating another shipment.

Operationally, the platform monitors:

• Oldest order-event age.
• Inventory-consumer failure rate.
• Carrier retry rate.
• Dead-letter count by event type.
• Shipment-processing duration.
• Backlog growth by queue.

During a carrier outage, shipment messages retry with backoff while inventory and invoice consumers continue independently. When the carrier recovers, shipment throughput is increased gradually to avoid another rate-limit incident.

Common Mistakes

Production Checklist

• Define delivery guarantees: document whether each workflow uses at-most-once or at-least-once delivery.
• Use stable message IDs: keep the same identifier across retries, redelivery, and replay.
• Make consumers idempotent: prevent duplicate business effects.
• Use database constraints: add unique keys for naturally unique operations.
• Deduplicate atomically: store processed IDs in the same transaction as business updates.
• Use a consistent envelope: include type, version, timestamp, producer, and identifiers.
• Separate commands and events: make message intent clear.
• Version schemas: support old messages during deployments and replay.
• Keep payloads small: store large files outside the broker.
• Validate at the boundary: reject malformed messages before side effects.
• Classify failures: distinguish transient, permanent, and unknown errors.
• Use exponential backoff: reduce pressure during dependency failures.
• Add jitter: prevent synchronised retry storms.
• Limit retries: define maximum attempts and message age.
• Configure dead-letter queues: isolate exhausted and poison messages.
• Monitor dead-letter traffic: assign clear operational ownership.
• Preserve original metadata: keep IDs, timestamps, source, and error details.
• Control concurrency: match worker capacity to downstream limits.
• Limit prefetch: avoid reserving more messages than workers can process.
• Apply backpressure: slow consumption when dependencies are overloaded.
• Set visibility timeouts carefully: exceed normal processing duration.
• Extend long-running leases: prevent premature redelivery.
• Acknowledge after success: avoid message loss before business work completes.
• Preserve only required ordering: avoid unnecessary global serialization.
• Choose balanced partition keys: reduce hot partitions.
• Use aggregate sequence numbers: detect stale or out-of-order messages.
• Design replay procedures: preserve IDs and track replay metadata.
• Throttle bulk replay: protect live traffic and recovered dependencies.
• Use a transactional outbox: avoid database-and-broker dual-write gaps.
• Protect dependencies: use timeouts, rate limits, bulkheads, and circuit breakers.
• Monitor queue depth: detect growing work volume.
• Monitor oldest-message age: measure real business delay.
• Monitor arrival and processing rates: compare demand with capacity.
• Monitor retry rates: identify dependency problems.
• Monitor processing duration: tune concurrency and visibility.
• Use structured logs: include message, queue, consumer, and correlation identifiers.
• Propagate tracing context: connect producer and consumer operations.
• Autoscale from queue signals: do not rely only on CPU.
• Respect dependency capacity: avoid scaling beyond database and API limits.
• Plan backlog recovery: prioritise live work and throttle historical traffic.
• Define retention: decide how long active and failed messages remain.
• Expire obsolete work: avoid processing messages that no longer have business value.
• Encrypt messages: protect data in transit and at rest.
• Use least privilege: separate producer, consumer, and operator permissions.
• Exclude secrets: never place credentials or access tokens in payloads.
• Audit replay operations: record who replayed which messages and why.
• Test duplicate delivery: prove that side effects happen once.
• Test consumer crashes: verify safe recovery before and after commits.
• Test broker outages: verify retries and backlog behaviour.
• Test poison messages: confirm they reach the correct failure path.
• Test schema compatibility: process old messages with new deployments.
• Test backlog recovery: measure whether dependencies remain stable.

Conclusion

Message queues improve resilience and scalability by separating producers from consumers and moving work out of synchronous request paths. However, reliable messaging depends on much more than publishing a payload and starting a worker.

Production systems should expect duplicate delivery, delayed messages, consumer crashes, broker outages, poison payloads, dependency rate limits, and large recovery backlogs. Stable message identifiers, idempotent consumers, bounded retries, dead-letter queues, visibility management, and controlled concurrency turn these failures into manageable operational conditions.

Ordering and partitioning should follow business requirements rather than being applied globally. Payloads should remain small, versioned, validated, and free of secrets. Producers that update databases should use the transactional outbox pattern to avoid losing events between independent systems.

Operational visibility is equally important. Queue depth, oldest-message age, processing rate, retries, dead-letter volume, and processing duration reveal whether work is progressing. Replay procedures, retention policies, security controls, and failure testing complete the production design.

Key Takeaway

A production message queue is reliable only when duplicate delivery is safe, failures are bounded and visible, consumers respect downstream capacity, and every important message can be traced, retried, isolated, and replayed without corrupting business state.

More Articles to Read

• Event-Driven Architecture in Distributed Systems
• Message Delivery Guarantees: At-Most-Once vs At-Least-Once vs Exactly-Once
• Background Workers Explained: Designing Reliable Asynchronous Processing
• Dead-Letter Queues, Retries, and Poison Messages
• Transactional Outbox Pattern for Reliable Messaging