Producers and Consumers

1The Producer's Job

A producer turns something that happened into a message and gets it durably into the broker. That sounds trivial; the details decide whether messages get lost, duplicated, or reordered.

// Kafka (kafkajs)
await producer.send({
  topic: "orders",
  acks: -1,                                  // wait for all in-sync replicas (09-delivery_guarantees/)
  messages: [{
    key: order.id,                           // same key → same partition → ordered per order
    value: JSON.stringify({ type: "OrderPlaced", orderId: order.id, total: order.total }),
    headers: { "event-id": crypto.randomUUID(), "schema-version": "3" },
  }],
});

A good message has:

2Producer Reliability Knobs

The dual-write problem applies to producers too: "commit DB, then publish" can lose the event if the process crashes between the two. The transactional outbox (backend/frameworks/16-background_jobs/ §6) fixes it.

3The Consumer's Job

A consumer receives messages, processes them, and tells the broker it's done (ack / commit offset). The order of those last two steps defines the delivery guarantee:

await consumer.run({
  eachMessage: async ({ topic, partition, message }) => {
    await handle(JSON.parse(message.value!.toString()));   // 1. process
    // 2. offset committed after this function resolves (autoCommit) — at-least-once
  },
});

• Process, then ack → a crash between them redelivers → at-least-once (handlers must be idempotent).
• Ack, then process → a crash between them loses the message → at-most-once.

09-delivery_guarantees/ covers this in depth.

4Push vs Pull, and Flow Control

• Push (RabbitMQ): the broker sends messages to the consumer as they arrive, limited by prefetch (basic.qos) — the max number of unacked messages per consumer. Prefetch 1 = fair dispatch but slow; prefetch 1000 = fast but one consumer can hoard work.
• Pull (Kafka, SQS): the consumer asks for the next batch when it's ready. Backpressure is natural: a busy consumer simply doesn't poll (distributed-systems/fundamentals/05-backpressure) — as long as it still polls often enough to stay in the group (max.poll.interval.ms).

5Scaling Consumers

• Competing consumers: run N instances on the same queue / in the same consumer group; throughput scales with N (Kafka: up to the partition count).
• Concurrency inside a consumer: process several messages in parallel — but only across different keys, or you lose per-key ordering (08-acknowledgements_and_message_ordering/).
• Poison messages: a message that always fails must not block the rest forever — retry a bounded number of times, then dead-letter it (10-dead_letter_queues/).

6Graceful Shutdown

On SIGTERM: stop fetching new messages, finish (or abandon and not ack) in-flight ones, commit offsets for what's done, leave the group cleanly so partitions are reassigned immediately rather than after a session timeout.

process.on("SIGTERM", async () => {
  await consumer.pause(consumer.assignment());   // 1. stop fetching
  await inFlight.drain();                        // 2. finish current messages
  await consumer.commitOffsets(processed);       // 3. commit what's done
  await consumer.disconnect();                   // 4. leave the group → fast rebalance
  process.exit(0);
});

Under the HoodHow Kafka's Idempotent Producer Deduplicates▾

When idempotence is enabled, the broker assigns the producer a producer ID (PID), and the producer numbers every record batch per partition with a monotonically increasing sequence number. The partition leader remembers, per PID, the last few sequence numbers it has appended:

• seq = last + 1 → append (normal case)
• seq ≤ last (a retry of something already written — the ack was lost) → acknowledge without appending again
• seq > last + 1 (a gap — something was lost or reordered) → reject with OutOfOrderSequence

That's what makes retries safe and keeps order with up to 5 in-flight requests. It only covers duplicates caused by the producer's own retries within one session; a producer that restarts and re-sends from application code gets a new PID — that's what transactional IDs (and consumer-side dedup) are for.

⚠️ Common Pitfalls

1. Fire-and-forget sends (acks=0, no publisher confirms) for data that matters.
2. Retries without idempotence → duplicates; max.in.flight > 1 without idempotence → reordering.
3. Committing offsets before processing → silent message loss on crash.
4. Non-idempotent handlers under at-least-once delivery → double charges, double emails.
5. Slow processing inside the poll loop → the consumer misses max.poll.interval.ms, gets kicked from the group, and triggers rebalances.
6. Huge payloads — put the blob in object storage and send a reference.

📋 Quick Reference

🎤 Interview Q&A

🛠️ Practice exercises

Each exercise is a file in this topic folder. Code exercises are self-checking: run them and make the checks pass.

✅ Check yourself

Source: broker_producers_and_consumers_theory.md · generated by tools/build_lessons.py