How does backpressure propagate in Reactor Kafka, and how is it implemented in terms of consumer pause/resume?

## What backpressure is In **Reactive Streams**, data flows from a Publisher to a Subscriber, but the *Subscriber controls the rate* via `Subscription.request(n)` — it asks for at most n items. A compliant Publisher must never emit more than the outstanding demand. This prevents a fast producer from flooding a slow consumer's memory. ## The Kafka mismatch Kafka's consumer **fetch protocol is push-batch oriented**: `consumer.poll()` returns whatever batches the broker has ready, bounded only by `max.poll.records` and fetch byte limits. There is no native 'give me exactly 3 records' pull. So Reactor Kafka must *bridge* demand-based backpressure onto a batch-fetch client. ## How Reactor Kafka bridges it: pause/resume The `KafkaConsumer` offers two flow-control methods: - `pause(Collection<TopicPartition>)` — subsequent `poll()` calls return **no records** for those partitions, but the consumer **keeps polling** (so it still heartbeats and stays in the group). - `resume(...)` — fetching restarts. Reactor Kafka's receive operator works like this: 1. It polls and feeds records into an internal queue that downstream operators drain according to their `request(n)` demand. 2. When downstream demand is satisfied and the queue has buffered enough (back-pressure threshold reached), it calls **consumer.pause()** on the assigned partitions — fetching stops, memory stays bounded. 3. As the downstream subscriber requests more and the queue drains, it calls **consumer.resume()** and polling delivers records again. Thus a slow `concatMap`/`flatMap` processing step automatically throttles broker ingestion — no manual code needed. ## Why heartbeats keep working Crucially, pause() does NOT stop the poll loop; it only stops *fetching records*. The consumer still calls poll() internally, which drives heartbeats and group membership. This is the safe way to slow down — contrast with simply *not calling poll()* in a plain client, which would blow `max.poll.interval.ms` and trigger a rebalance. ## Failure modes if backpressure is mishandled - **OOM:** if you break the demand chain (e.g. an operator that requests Long.MAX_VALUE unboundedly and buffers), records accumulate without bound. - **Rebalance storms / 'kicked from group':** if processing per poll exceeds `max.poll.interval.ms`, the broker assumes the consumer is dead and reassigns its partitions. Reactor Kafka mitigates by pausing rather than stalling the loop, but slow processing of *already-fetched* records can still exceed the interval if not offloaded. ## Tuning levers - `max.poll.records` — fewer records per poll = finer-grained, lower memory. - `ReceiverOptions` poll timeout and commit settings. - Reactor `prefetch`/`limitRate(n)` to cap demand. - Bounded `flatMap(..., concurrency)` so in-flight work is capped. ## Vert.x parallel The Vert.x Kafka consumer exposes the same idea explicitly via `consumer.pause()`/`consumer.resume()` and a `fetch(amount)` demand API on its read stream — same underlying mechanism, different surface.