What is the request purgatory in a Kafka broker, and why does it exist?

## The problem A Kafka broker serves client requests using a small, fixed pool of **request-handler threads** (`num.io.threads`, default 8). Some requests cannot be answered the instant they arrive: - A **produce** request with `acks=all` (or `acks=-1`) must not be acknowledged until every **in-sync replica (ISR)** has copied the new records. That replication takes time and depends on follower brokers fetching. - A **fetch** request from a consumer may set `fetch.min.bytes` (default 1). If the broker doesn't yet have that many bytes of new data, the consumer would rather wait a little than get an empty response — up to `fetch.max.wait.ms` (default 500 ms). This is the long-poll behavior. If the handler thread simply *blocked* waiting for these conditions, the broker would exhaust its thread pool almost immediately under load, since thousands of fetches and replications can be outstanding at once. ## The solution: purgatory + DelayedOperation Kafka decouples *waiting* from *threads*. When a request can't complete immediately, the broker wraps it in a **`DelayedOperation`** subclass: - **`DelayedProduce`** — for `acks=all` produce requests. - **`DelayedFetch`** — for fetches waiting on `fetch.min.bytes` / `fetch.max.wait.ms`. - Others exist too (`DelayedJoin`, `DelayedHeartbeat`, `DelayedDeleteRecords`, etc.). The operation is placed into a **`DelayedOperationPurgatory`**. The handler thread is then *released* to serve other work. Each operation defines: - `tryComplete()` — checks whether its condition is now satisfied (e.g. has the high watermark advanced past the produced offset for all partitions?). - `onComplete()` — sends the response back to the client. - `onExpiration()` — what to do if the timeout fires first (e.g. respond to the producer with whatever acks were collected, or send an empty fetch response). ## Two ways an operation finishes 1. **Event-driven completion.** When something changes that *might* satisfy waiting operations — most importantly when a partition's **high watermark (HW)** or **ISR** advances after followers replicate — the broker calls `checkAndComplete(key)` on the purgatory for the affected **watcher key** (usually a `TopicPartition`). That re-runs `tryComplete()` on every operation watching that key; any whose condition is now true are completed and removed. 2. **Timeout.** Every operation is also handed to a timer with its deadline. If the deadline passes first, the operation is force-completed via `onExpiration()`. ## Why it scales Because waiting is represented as *data* (parked operations + watcher lists + timer entries) rather than *blocked threads*, a broker can keep millions of consumer long-polls and replication waits in flight with only a handful of threads. The timer is a **hierarchical timing wheel**, which adds/removes timeouts in O(1), so even a flood of short-lived waits is cheap. ## Edge cases / nuances - An operation can watch **multiple keys** (a produce to N partitions watches N `TopicPartition` keys); it completes once, the first time *all* its conditions are met. - `tryComplete()` may be called many times (once per triggering event) before it finally succeeds — it must be idempotent and cheap. - Completion is guarded so an operation is completed exactly once even if a timeout and an event race. - Purgatory has a **reaper thread** that periodically purges already-completed operations still lingering in watcher lists (controlled by `purge.interval.requests`), preventing memory bloat.