How message brokers actually deliver

HOW IT FITS TOGETHER

KEY TERMS

At-most / at-least / exactly-once

Ack-then-process can lose work; process-then-ack can repeat it. Brokers that say exactly-once mean a scoped guarantee, not a property of your HTTP calls or emails.

Ack and lease

SQS visibility timeout (default 30 s, max 12 h), JetStream AckWait, Pub/Sub ack deadline, RabbitMQ channel lifetime, Redis Streams pending entry list. Long jobs must extend the lease or expect a second worker.

Consumer groups and partitions

Kafka assigns each partition to one consumer in a group, so partition count caps parallelism. Queues (SQS, RabbitMQ) instead compete per message, which spreads load but drops ordering.

Ordering scope

Order holds per Kafka partition (key), SQS FIFO message group, or Pub/Sub ordering key. Retries, DLQs, and parallel consumers all break wider ordering assumptions.

Dedupe windows

SQS FIFO drops resends with the same deduplication ID for 5 minutes. Kafka's idempotent producer stops retry duplicates in the log; its transactions plus read_committed give exactly-once only for Kafka-to-Kafka processing.

Dead-letter queue

Where a message goes after N failed deliveries (SQS maxReceiveCount, quorum queue delivery-limit, JetStream MaxDeliver). Without one, a poison message blocks or loops forever.

Transactional outbox

Write the event to an outbox table in the same transaction as the state change, then relay it. It removes lost or phantom events, not duplicates, so consumers stay idempotent.

IN YOUR STACK

import { SQSClient, ReceiveMessageCommand, DeleteMessageCommand } from "@aws-sdk/client-sqs";

const sqs = new SQSClient({});
const QueueUrl = process.env.QUEUE_URL!;

while (true) {
  const { Messages = [] } = await sqs.send(new ReceiveMessageCommand({
    QueueUrl,
    MaxNumberOfMessages: 10,
    WaitTimeSeconds: 20,      // long polling
    VisibilityTimeout: 60,    // lease: must exceed worst-case handling time
    MessageSystemAttributeNames: ["ApproximateReceiveCount"],
  }));
  for (const msg of Messages) {
    const event = JSON.parse(msg.Body!);
    await handleOnce(event.eventId, event);   // idempotent by business key
    // The ack: until this succeeds, the message comes back after the timeout.
    await sqs.send(new DeleteMessageCommand({ QueueUrl, ReceiptHandle: msg.ReceiptHandle! }));
  }
}

import amqp from "amqplib";

const conn = await amqp.connect(process.env.AMQP_URL!);
const ch = await conn.createChannel();
await ch.assertExchange("orders.dlx", "fanout", { durable: true });
await ch.assertQueue("orders", {
  durable: true,
  arguments: { "x-queue-type": "quorum", "x-dead-letter-exchange": "orders.dlx" },
});
await ch.prefetch(50); // max unacked deliveries in flight on this channel

await ch.consume("orders", async (msg) => {
  if (!msg) return; // consumer was cancelled by the broker
  try {
    await handleOnce(JSON.parse(msg.content.toString()), msg.fields.redelivered);
    ch.ack(msg);
  } catch {
    ch.nack(msg, false, false); // requeue=false: dead-letter instead of hot-looping
  }
}, { noAck: false });

Properties props = new Properties();
props.setProperty("bootstrap.servers", "localhost:9092");
props.setProperty("group.id", "billing");
props.setProperty("enable.auto.commit", "false");
props.setProperty("isolation.level", "read_committed"); // skip aborted txn writes
props.setProperty("key.deserializer", StringDeserializer.class.getName());
props.setProperty("value.deserializer", StringDeserializer.class.getName());

try (var consumer = new KafkaConsumer<String, String>(props)) {
  consumer.subscribe(List.of("orders"));
  while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(500));
    for (ConsumerRecord<String, String> r : records) {
      // Same key -> same partition -> ordered for that key only.
      handleOnce(r.topic(), r.partition(), r.offset(), r.value());
    }
    consumer.commitSync(); // commit after processing = at-least-once
  }
}

import redis

r = redis.Redis(decode_responses=True)
STREAM, GROUP, ME = "orders", "billing", "worker-1"
try:
    r.xgroup_create(STREAM, GROUP, id="0", mkstream=True)
except redis.exceptions.ResponseError:
    pass  # BUSYGROUP: the group already exists

while True:
    # Take over entries another consumer read but never acked (idle > 60 s).
    _, claimed, _deleted = r.xautoclaim(STREAM, GROUP, ME, min_idle_time=60_000,
                                        start_id="0-0", count=50)
    fresh = r.xreadgroup(GROUP, ME, {STREAM: ">"}, count=50, block=5_000)
    entries = claimed + [e for _stream, batch in fresh for e in batch]
    for entry_id, fields in entries:
        handle_once(fields["eventId"], fields)
        r.xack(STREAM, GROUP, entry_id)

import { connect, nanos } from "@nats-io/transport-node";
import { jetstream, jetstreamManager, AckPolicy } from "@nats-io/jetstream";

const nc = await connect({ servers: "nats://localhost:4222" });
const jsm = await jetstreamManager(nc);
await jsm.consumers.add("ORDERS", {
  durable_name: "billing",
  ack_policy: AckPolicy.Explicit,
  ack_wait: nanos(30_000), // the lease, in nanoseconds
  max_deliver: 5,          // default -1: redeliver forever
});

const consumer = await jetstream(nc).consumers.get("ORDERS", "billing");
for await (const m of await consumer.consume({ max_messages: 50 })) {
  try {
    await handleOnce(m.json(), m.info.deliveryCount);
    m.ack();
  } catch (err) {
    if (m.info.deliveryCount >= 5) m.term(); // stop redelivery
    else m.nak(5_000);                       // retry after 5 s
  }
}

from google.cloud import pubsub_v1
from google.cloud.pubsub_v1.subscriber import exceptions as sub_exceptions

subscriber = pubsub_v1.SubscriberClient()
# Pull subscription created with enable_exactly_once_delivery=True.
path = subscriber.subscription_path("my-project", "orders-billing")

def callback(message: pubsub_v1.subscriber.message.Message) -> None:
    handle_once(message.attributes["event_id"], message.data)
    try:
        # Only a successful ack future guarantees no redelivery.
        message.ack_with_response().result()
    except sub_exceptions.AcknowledgeError as e:
        log.warning("ack failed (%s); expect redelivery", e.error_code)

flow = pubsub_v1.types.FlowControl(max_messages=100)
future = subscriber.subscribe(path, callback=callback, flow_control=flow)
with subscriber:
    future.result()

WHERE IT BITES

• Acking before the side effect commits (auto-ack, auto-commit, noAck) turns every crash into silent data loss.
• Using the broker's message ID as the idempotency key: producer retries and outbox relays create new IDs for the same event. Dedupe on a business event ID stored with the side effect.
• A lease shorter than the slowest handler, or a GC pause, makes two workers process the same message concurrently; extend the lease or use a DB unique constraint.
• Requeueing a poison message immediately with no delivery limit burns CPU and blocks ordered groups; cap attempts and dead-letter.
• Assuming global order: parallel consumers, retries, and DLQ redrives reorder events. Carry a version or sequence and reject stale updates.
• Treating a DLQ as a bin: without alerts on depth and a tested redrive path, failures just disappear more slowly.

WHEN IT BREAKS IN PRODUCTION

• One bad message blocks or floods the queue
• A brief blip turns into a long outage
• Webhooks arrive twice, late, or out of order
• A retry creates two payments

SOURCES

• Amazon SQS visibility timeout (opens in new tab)Checked 2026-10-07
• Exactly-once processing in Amazon SQS (FIFO) (opens in new tab)Checked 2026-10-07
• RabbitMQ: Consumer acknowledgements and publisher confirms (opens in new tab)Checked 2026-10-07
• Apache Kafka design: message delivery semantics (opens in new tab)Checked 2026-10-07
• Pub/Sub exactly-once delivery (opens in new tab)Checked 2026-10-07
• Pattern: Transactional outbox (opens in new tab)Checked 2026-10-07