# Producers and Consumers

## 1. The 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.

```ts
// 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:

| Part | Purpose |
|---|---|
| **key** | partitioning / ordering scope (`orderId`, `userId`) |
| **event id** | lets consumers deduplicate redeliveries |
| **type + schema version** | consumers can evolve independently |
| **timestamp** | when it *happened* (not when it was consumed) |
| **payload** | small; IDs + the facts consumers need, not whole aggregates |

## 2. Producer Reliability Knobs

| Setting | Effect |
|---|---|
| `acks` (Kafka) / publisher confirms (RabbitMQ) | does `send()` resolve only after the broker has durably stored it? |
| retries | resend on transient errors — can **duplicate** unless the producer is idempotent |
| idempotent producer (`enable.idempotence=true`) | broker drops duplicates using producer id + sequence numbers |
| `max.in.flight.requests` | >1 with retries and no idempotence can **reorder** messages |
| batching (`linger.ms`, `batch.size`) | trade a few ms of latency for much higher throughput |
| compression | smaller network/disk footprint for batched payloads |

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.

## 3. The 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:

```ts
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.

## 4. Push 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`).

## 5. Scaling 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/`).

## 6. Graceful 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.

## Under the Hood: How 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.

![A producer adds a key, event id and schema version, sends with acks and an idempotent sequence number; a consumer processes then commits its offset, with retries ending in a dead-letter queue](diagrams/01-producer-consumer-contract.svg)

## 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

| Producer | Consumer |
|---|---|
| key = ordering scope | process → then ack/commit |
| `acks=all` + idempotence | idempotent handler (dedup by event id) |
| outbox for DB + publish | bounded retries → DLQ |
| batch + compress for throughput | prefetch / poll size for flow control |
| event id, type, schema version | graceful shutdown: stop, finish, commit, leave |