Queues and Streams

Decoupling producers from consumers: how a work queue differs from a log, what consumer groups do, and what replay buys you.

SyntaxReceiveMessageCommand({ QueueUrl, MaxNumberOfMessages: 10, WaitTimeSeconds: 20, VisibilityTimeout: 60 })

A queue lets a producer hand off work and return immediately. The producer's latency stops depending on the consumer's, a burst becomes a backlog instead of a wall of errors, and the consumer can be redeployed, slowed or scaled without the caller noticing. That is the whole value: the two sides no longer have to be available or fast at the same moment.

A work queue is not a log

In a work queue — SQS, RabbitMQ — a message is a task owned by one consumer, acknowledged, then gone. Delivery is at-least-once, so handlers must be idempotent, and ordering is a special mode rather than the default. In a log — Kafka, Kinesis — a record is appended to a partition and kept for a retention period whether anyone read it or not. Consumers hold a position rather than removing anything, so many independent readers can consume the same record for different purposes.

Consumer groups

A Kafka consumer group shares out a topic's partitions, one partition to at most one member, so the group's parallelism is capped by partition count and ordering is preserved within a partition. Separate groups get full independent copies of the stream. Share groups (KIP-932) give a log queue-like semantics — cooperative consumption and per-record acknowledgement, without the partition-per-consumer limit; they shipped as early access in Kafka 4.0, as a preview in 4.1, and became generally available in 4.2.

What replay buys you

Because a log keeps records, a consumer can be rewound. A new service can be backfilled from history, a bad deploy's output can be recomputed, and a derived store — a cache, a search index, a read model — can be rebuilt from the stream rather than migrated.

Example

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

const sqs = new SQSClient({});
const QueueUrl = process.env.QUEUE_URL;
let running = true;
process.on("SIGTERM", () => { running = false; });

while (running) {
  const { Messages = [] } = await sqs.send(new ReceiveMessageCommand({
    QueueUrl,
    MaxNumberOfMessages: 10, // 10 is the maximum per receive
    WaitTimeSeconds: 20,     // long poll. 0, the default, is a busy loop that
                             // bills for empty receives and adds latency.
    VisibilityTimeout: 60,   // overrides the queue default of 30s (max 12h)
  }));

  for (const m of Messages) {
    // Heartbeat: push the deadline out while we are still working, or SQS
    // redelivers this message to another worker mid-flight.
    const beat = setInterval(() => {
      sqs.send(new ChangeMessageVisibilityCommand({
        QueueUrl, ReceiptHandle: m.ReceiptHandle, VisibilityTimeout: 60,
      })).catch(() => {});
    }, 30_000);

    try {
      // At-least-once delivery: this must be safe to run twice. Key the work
      // on something stable, not on a receive count.
      await handle(JSON.parse(m.Body));
      // Deleting is the acknowledgement. Nothing before this point is final.
      await sqs.send(new DeleteMessageCommand({
        QueueUrl, ReceiptHandle: m.ReceiptHandle,
      }));
    } catch (err) {
      // Do not delete. After maxReceiveCount redeliveries the queue's redrive
      // policy moves it to the dead-letter queue instead of looping forever.
      log.error({ err, messageId: m.MessageId }, "handler failed");
    } finally {
      clearInterval(beat);
    }
  }
}

When to use it

  • Taking image processing and email out of a request handler so the endpoint returns in milliseconds.
  • Deciding between SQS and Kafka when three teams want to react to the same order event.
  • Rebuilding a search index or cache from history after fixing a bug in the indexer.
  • Explaining why a slow job is running twice and why the fix is idempotency rather than a longer timeout.

More examples

A Kafka consumer group

Use a group per concern, not per instance: invoicing, search indexing and analytics each read the orders topic independently and cannot block one another. Scaling beyond the partition count does nothing, since a partition is assigned to at most one member of a group — partition count is the parallelism ceiling you choose when you create the topic.

Example · javascript
import { Kafka } from "kafkajs";

const kafka = new Kafka({ clientId: "billing", brokers: ["b1:9092"] });

// groupId IS the subscription. Every instance sharing this id splits the
// partitions between them; a second, different groupId gets its own full copy
// of the same records.
const consumer = kafka.consumer({ groupId: "billing-invoicer" });

await consumer.connect();
// Current spelling takes an array:
await consumer.subscribe({ topics: ["orders"], fromBeginning: false });
// The older single-topic spelling is deprecated but still seen:
// await consumer.subscribe({ topic: "orders", fromBeginning: false });

await consumer.run({
  autoCommit: false, // commit only after the work is durable
  eachMessage: async ({ topic, partition, message, heartbeat }) => {
    await invoice(JSON.parse(message.value.toString()));
    await heartbeat();
    // Commit the NEXT offset to read, which is this one plus one.
    await consumer.commitOffsets([
      { topic, partition, offset: (Number(message.offset) + 1).toString() },
    ]);
  },
});

Retention and compaction: how long a log remembers

Retention is the budget for replay: a consumer can only be rewound as far back as the log still goes, so the retention setting and the recovery story are the same decision. Compaction is what you want for state-shaped topics, where a consumer needs the current value of every key rather than every change.

Example · bash
# Keep 30 days instead of the 7-day default (retention.ms = 604800000),
# because a new consumer should be able to backfill a month of history.
kafka-configs.sh --bootstrap-server b1:9092 --alter \
  --entity-type topics --entity-name orders \
  --add-config retention.ms=2592000000

# A compacted topic is a different thing: cleanup.policy defaults to delete,
# but 'compact' keeps the LATEST record per key forever and drops older ones.
# That makes the topic a replayable snapshot of current state, not a history.
kafka-configs.sh --bootstrap-server b1:9092 --alter \
  --entity-type topics --entity-name customer-profiles \
  --add-config cleanup.policy=compact

# A tombstone - a record with a null value - is how a key is deleted.

Replaying to rebuild a derived store

This is the capability a work queue cannot give you, and the reason to choose a log when a downstream view may need rebuilding: fix the indexer, reset the group, replay. It also sets a design constraint — replayed records arrive again in full, so the consumer has to be idempotent or write with a version check.

Example · bash
# Rebuilding a search index or a read model: move a group's offsets, then
# start the consumers and let them re-process history.

# The group must be stopped - Kafka refuses to reset an active group.
kafka-consumer-groups.sh --bootstrap-server b1:9092 \
  --group search-indexer --describe   # note CURRENT-OFFSET and LAG first

# Dry run prints the target offsets and changes nothing.
kafka-consumer-groups.sh --bootstrap-server b1:9092 \
  --group search-indexer --topic orders \
  --reset-offsets --to-datetime 2026-09-01T00:00:00.000 --dry-run

# Same command with --execute applies it.
kafka-consumer-groups.sh --bootstrap-server b1:9092 \
  --group search-indexer --topic orders \
  --reset-offsets --to-earliest --execute

Related

sd idempotencysd backpressuresd timeouts retriessd searchsd design notifications

Discussion

  • Be the first to comment on this lesson.