feat: pause Kafka partitions instead of blocking backoff for non-keyed messages

docs/adr/01-kafka-retry-backoff.md

@@ -0,0 +1,118 @@
+# Kafka Retry Backoff for Non-Keyed Topics
+
+## 1. Context
+
+`KafkaSubscriber`'s retry design (`_RetryMiddleware`, exponential backoff via
+`NACK_ON_ERROR` plus an inline `asyncio.sleep`) blocks the whole poll loop for
+that consumer while a message backs off. Every partition the consumer owns
+stalls, not just the failing one.
+
+For a keyed topic that is fine: a partition is already one logical ordered
+stream, so blocking it while a message backs off does not hold up anything
+that was not already serialized behind it.
+
+For a non-keyed topic (no ordering requirement), the same partition carries
+unrelated messages round-robined onto it. Blocking it delays those messages
+for no ordering reason.
+
+Four options were considered.
+
+## 2. Options considered
+
+### 2.1 Numbered retry topics
+
+Java's `@RetryableTopic`, or the same idea as Rabbit's per-delay dead-letter
+queues: give each backoff tier its own topic (`topic.5`, `topic.30`, ...).
+
+Rejected. Kafka has no message TTL or broker-managed delayed delivery like
+Rabbit's `x-message-ttl` plus DLX, so every tier still needs the same
+in-process sleep this design already has. The only thing gained is isolating
+the block to one topic. Rabbit's DLX queues are declared by the app at
+runtime and expire themselves (`x-expires`); Kafka topics are
+ops-provisioned (partitions, replication, retention, monitoring), so N new
+topics per subscriber is real infrastructure work for a small isolation win.
+
+### 2.2 External delay store plus poller
+
+Ack the message immediately on failure, write `{payload, due_at, attempt}`
+to Redis or a Mongo collection, and have a background poller republish it to
+the original topic once due.
+
+Rejected. Adds a second moving part (a poller, its own failure modes, its
+own monitoring) for state the in-process fetch loop already tracks
+(`_retry_state`). Too much for the size of the actual problem.
+
+### 2.3 `max_workers` concurrency
+
+FastStream's confluent broker supports running several handlers
+concurrently off one shared semaphore, so unrelated messages on the same
+partition could process while one backs off.
+
+Rejected, for two independent reasons.
+
+1. No partition affinity: dispatch is a flat semaphore with no per-partition
+   reservation, so a single hot partition can consume every worker slot
+   itself.
+2. FastStream requires `ack_policy=ACK_FIRST` whenever `max_workers>1`
+   (raises `SetupError` otherwise), committing the offset before the
+   handler runs. That breaks the NACK-and-redeliver retry this whole design
+   depends on. Moving the retry loop into the handler itself works
+   mechanically, but the retry state then lives only in process memory: a
+   crash mid-retry loses the message outright, since the offset is already
+   committed and Kafka will not redeliver it. Confirmed against a real
+   broker: a `SIGKILL` mid-retry loses the message.
+
+### 2.4 Partition pause and resume
+
+`confluent_kafka.Consumer.pause()` / `.resume()`: on failure, pause the
+failing partition and schedule a background resume after the backoff delay,
+instead of blocking the poll loop with `asyncio.sleep`.
+
+Chosen. See Decision.
+
+## 3. Decision
+
+Use partition pause and resume, gated to messages with no Kafka key
+(`raw_message.key() is None`). Keyed messages keep today's inline
+`asyncio.sleep` unchanged: their partition is already a single ordered
+stream, so blocking it costs nothing extra, and pausing it would add
+complexity for no benefit.
+
+For a non-keyed message: pause the specific `(topic, partition)`, keep
+polling (other partitions the consumer owns keep flowing since the fetch
+loop no longer awaits a sleep), and resume that partition once the backoff
+delay elapses. The offset is never committed during this, so redelivery and
+attempt counting are unchanged from today.
+
+Verified against a real broker that `pause()` is purely client-side: a
+`SIGKILL` mid-pause leaves nothing durable. A fresh consumer instance
+assigned the same partition reads normally right away, no stuck state.
+
+## 4. Consequences
+
+- Unrelated messages on a non-keyed topic's other partitions no longer wait
+  on one partition's backoff.
+- Within the same partition, order is still strict FIFO: a backing-off
+  message still blocks whatever is queued directly behind it on that
+  partition. Kafka cannot skip a message and return to it later without
+  external state (rejected in 2.2), so this is an accepted limit, not a bug.
+- Pause and resume calls run through the same single-thread executor
+  FastStream already uses to serialize all confluent-kafka client calls
+  (the client is not thread-safe). This is why the call cannot just happen
+  inline on the event loop.
+- No ack-policy change, no new infrastructure, no loss of at-least-once
+  delivery.
+
+## 5. Implementation notes
+
+- `message.consumer` is typed `ConsumerProtocol` (only `commit()`/`seek()`);
+  pause/resume reach past it to the concrete `AsyncConfluentConsumer`
+  FastStream actually constructs, via `# type: ignore[attr-defined]`.
+- A batch mixing keyed and non-keyed records is classified `any()`-wise:
+  one keyed record in the batch is enough to take the safe (blocking) path,
+  since a batch is acked/committed as one unit.
+- `pause()` failing must not propagate out of `_backoff` - it would replace
+  the caller's original exception instead of letting it re-raise for
+  redelivery. Falls back to inline `asyncio.sleep` on that failure.
+- `resume()` retries a few times with backoff before giving up, instead of
+  stranding a partition paused forever on the first failure.

docs/signals.md

@@ -158,13 +158,17 @@ order_publisher = KafkaSubscriber.router.publisher("my_service.order.create")
 
 Everything FastStream's confluent router supports — `batch`, `ack_policy`, multiple topics per subscriber, etc. — is available directly; fastloom doesn't wrap it.
 
-`KafkaSubscriber(settings, base_delay=5, max_delay=240, exceptions=None, ack_policy=None, allow_auto_create_topics=True, acks=1, enable_idempotence=False)` applies an exponential-backoff-with-jitter `asyncio.sleep` on exception, throttling `NACK_ON_ERROR` redelivery instead of the DLX-queue chain Rabbit uses (Kafka has no per-message TTL primitive to build one from). This is a **broker-level** middleware — it wraps every subscriber on `KafkaSubscriber.router`, not an opt-in per `@subscriber(...)` call like Rabbit's `retry_backoff=`. It also sets `NACK_ON_ERROR` as the broker's default `ack_policy` (reaching into `router.broker.config.broker_config.ack_policy` — the one mutable field the read-only composed `broker.config.ack_policy` property actually reads from), so redelivery works out of the box; pass `ack_policy=` to pick a different broker-wide default, or set `ack_policy=` on an individual `@subscriber(...)` call to override just that one. `enable_idempotence=True` forces `acks="all"` regardless of the `acks` param — librdkafka itself rejects `enable.idempotence` with any other `acks` value at producer construction (verified directly against the installed `confluent_kafka.Producer`), so this is resolved for you rather than left as a footgun.
+`KafkaSubscriber(settings, base_delay=5, max_delay=240, exceptions=None, ack_policy=None, allow_auto_create_topics=True, acks=1, enable_idempotence=False)` retries failed messages via `NACK_ON_ERROR` redelivery with exponential backoff, instead of the DLX-queue chain Rabbit uses (Kafka has no per-message TTL primitive to build one from). This is a **broker-level** middleware — it wraps every subscriber on `KafkaSubscriber.router`, not an opt-in per `@subscriber(...)` call like Rabbit's `retry_backoff=`. It also sets `NACK_ON_ERROR` as the broker's default `ack_policy` (reaching into `router.broker.config.broker_config.ack_policy` — the one mutable field the read-only composed `broker.config.ack_policy` property actually reads from), so redelivery works out of the box; pass `ack_policy=` to pick a different broker-wide default, or set `ack_policy=` on an individual `@subscriber(...)` call to override just that one. `enable_idempotence=True` forces `acks="all"` regardless of the `acks` param — librdkafka itself rejects `enable.idempotence` with any other `acks` value at producer construction (verified directly against the installed `confluent_kafka.Producer`), so this is resolved for you rather than left as a footgun.
+
+How the backoff actually waits depends on whether the message carries a Kafka key:
+
+- **Keyed** (`key` set): blocks that subscriber's whole poll loop with `asyncio.sleep` before retrying — every partition/topic it owns stalls, not just the failing one. `max_delay` must therefore stay under whatever `max.poll.interval.ms` is configured for that consumer (FastStream's own default is 5 minutes), or the broker's group coordinator decides the consumer is dead and triggers a rebalance mid-backoff. The default `max_delay=240` (4 minutes) leaves a minute of margin; raise both together if you need longer backoff.
+- **Non-keyed** (`key is None`): pauses only the failing `(topic, partition)` via `confluent_kafka.Consumer.pause()`/`.resume()` (run through the same single-thread executor FastStream uses to serialize all consumer calls, since the client isn't thread-safe) and keeps polling — sibling partitions stay unaffected. Within that one partition, order is still strict FIFO, so a backing-off message still blocks whatever's queued directly behind it there.
 
 Things to know before relying on it:
 
 - A subscriber that deliberately overrides back to `ACK_FIRST` (offset commits before the handler runs) just gets its backoff silently skipped — the original exception still propagates untouched, since there's no way to opt a single subscriber out of this broker-wide middleware otherwise.
-- The sleep blocks that subscriber's whole poll loop — **every** partition/topic it owns, not just the failing one — since the loop can't call `poll()` again until the current message's handler (and our sleep) returns. `max_delay` must therefore stay under whatever `max.poll.interval.ms` is configured for that consumer (FastStream's own default is 5 minutes), or the broker's group coordinator decides the consumer is dead and triggers a rebalance mid-backoff — worse than the original poison-message problem. The default `max_delay=240` (4 minutes) leaves a minute of margin under that 5-minute default; raise both together if you need longer backoff.
-- There's no safe way to isolate one stuck partition from a subscriber's other partitions today. `max_workers>1` looks like the fix but isn't: `KafkaMessage.ack()` commits the consumer's *current* position, not a specific offset, so a later offset's success can commit past an earlier offset that's still asleep in backoff — a crash in that window permanently skips the earlier message. Don't reach for `max_workers` as a mitigation for this until it's fixed upstream or reworked here.
+- See [docs/adr/01-kafka-retry-backoff.md](adr/01-kafka-retry-backoff.md) for why the keyed/non-keyed split exists, and why numbered retry topics, an external delay store, and `max_workers` concurrency were all considered and rejected in favor of pause/resume.
 
 `auto_offset_reset` has no broker-level equivalent — unlike `ack_policy`, it isn't composed from a shared config object, it flows straight from each `@subscriber(...)` call into the raw confluent-kafka consumer config. Pass it per-subscriber (see the example above).
 
@@ -197,3 +201,4 @@ There's no Kafka equivalent of `RabbitPayloadTelemetryMiddleware` — Kafka span
 - [db.md](db.md) — `BaseDocumentSignal` auto-publishes to the same broker.
 - [Observability](observability.md) — Rabbit/Kafka instrumentation and queue-name filtering.
 - [Healthcheck](healthcheck.md) — broker ping registration.
+- [docs/adr/01-kafka-retry-backoff.md](adr/01-kafka-retry-backoff.md) — why Kafka backoff branches on keyed vs. non-keyed messages.

fastloom/signals/kafka/depends.py

@@ -19,9 +19,10 @@
 from fastloom.utils import exponential_backoff
 
 if TYPE_CHECKING:
+    from confluent_kafka import Message
     from faststream._internal.types import BrokerMiddleware
     from faststream.confluent.fastapi import KafkaRouter
-    from faststream.confluent.message import KafkaMessage
+    from faststream.confluent.message import ConsumerProtocol, KafkaMessage
     from faststream.confluent.parser import AsyncConfluentParser
     from faststream.confluent.publisher.producer import (
         AsyncConfluentFastProducerImpl,
@@ -288,6 +289,7 @@ class KafkaSubscriber(SelfSustaining):
     _max_delay: int
     _exceptions: tuple[type[Exception], ...]
     _retry_state: dict[tuple[str, int], _RetryState]
+    _pending_resumes: set[asyncio.Task[None]]
 
     def __init__(
         self,
@@ -310,6 +312,7 @@ def __init__(
         self._max_delay = max_delay
         self._exceptions = tuple(exceptions or [Exception])
         self._retry_state = {}
+        self._pending_resumes = set()
         subscriber = self
 
         class _RetryMiddleware(BaseMiddleware):
@@ -322,7 +325,7 @@ async def consume_scope(self, call_next, msg):
                     result = await call_next(msg)
                 except subscriber._exceptions:
                     if msg.is_manual:
-                        await subscriber._backoff(key)
+                        await subscriber._backoff(key, msg)
                     raise
                 else:
                     subscriber._clear_retry_state(key)
@@ -343,14 +346,27 @@ async def consume_scope(self, call_next, msg):
             ack_policy if ack_policy is not None else AckPolicy.NACK_ON_ERROR
         )
 
+        original_stop = self.router.broker.stop
+
+        async def _stop_and_cancel_pending_resumes(*args, **kwargs):
+            for task in list(self._pending_resumes):
+                task.cancel()
+            await original_stop(*args, **kwargs)
+
+        self.router.broker.stop = _stop_and_cancel_pending_resumes
+
     @staticmethod
-    def _locate(message: KafkaMessage) -> _MessageKey | None:
+    def _first_record(message: KafkaMessage) -> Message:
         raw = message.raw_message
         # batch messages are a tuple against a real broker but a list
         # against FastStream's own confluent test/mock broker - a single
         # confluent_kafka.Message is never a Sequence, so this only ever
         # takes the batch branch.
-        first = raw[0] if isinstance(raw, Sequence) else raw
+        return raw[0] if isinstance(raw, Sequence) else raw
+
+    @classmethod
+    def _locate(cls, message: KafkaMessage) -> _MessageKey | None:
+        first = cls._first_record(message)
         topic, partition, offset = (
             first.topic(),
             first.partition(),
@@ -366,7 +382,14 @@ def _clear_retry_state(self, key: _MessageKey) -> None:
         if last is not None and last.offset == key.offset:
             del self._retry_state[partition_key]
 
-    async def _backoff(self, key: _MessageKey) -> None:
+    @classmethod
+    def _is_keyed(cls, message: KafkaMessage) -> bool:
+        raw = message.raw_message
+        # mixed-keyedness batch defaults to the safe path - docs/adr/01
+        records = raw if isinstance(raw, Sequence) else (raw,)
+        return any(record.key() is not None for record in records)
+
+    async def _backoff(self, key: _MessageKey, message: KafkaMessage) -> None:
         partition_key = key.partition_key
         last = self._retry_state.get(partition_key)
         attempt = (
@@ -385,4 +408,78 @@ async def _backoff(self, key: _MessageKey) -> None:
             delay,
             attempt,
         )
+
+        if self._is_keyed(message):
+            # keyed: blocking costs nothing extra - see docs/adr/01
+            await asyncio.sleep(delay)
+            return
+
+        consumer = message.consumer
+        try:
+            await self._set_partition_paused(key, consumer, paused=True)
+        except Exception:
+            # must not propagate - would replace the caller's exception,
+            # see docs/adr/01.
+            logger.warning(
+                "failed to pause %s[%s] - falling back to inline backoff",
+                key.topic,
+                key.partition,
+                exc_info=True,
+            )
+            await asyncio.sleep(delay)
+            return
+
+        self._schedule_resume(key, consumer, delay)
+
+    async def _set_partition_paused(
+        self,
+        key: _MessageKey,
+        consumer: ConsumerProtocol,
+        *,
+        paused: bool,
+    ) -> None:
+        from confluent_kafka import TopicPartition
+        from faststream._internal.utils.functions import run_in_executor
+
+        # not thread-safe, reaches past ConsumerProtocol - see docs/adr/01
+        raw_consumer = consumer.consumer  # type: ignore[attr-defined]
+        op = raw_consumer.pause if paused else raw_consumer.resume
+        await run_in_executor(
+            consumer._thread_pool,  # type: ignore[attr-defined]
+            op,
+            [TopicPartition(key.topic, key.partition)],
+        )
+
+    def _schedule_resume(
+        self, key: _MessageKey, consumer: ConsumerProtocol, delay: float
+    ) -> None:
+        task = asyncio.create_task(
+            self._resume_partition(key, consumer, delay)
+        )
+        self._pending_resumes.add(task)
+        task.add_done_callback(self._pending_resumes.discard)
+
+    async def _resume_partition(
+        self, key: _MessageKey, consumer: ConsumerProtocol, delay: float
+    ) -> None:
         await asyncio.sleep(delay)
+        for attempt in range(1, 4):
+            try:
+                await self._set_partition_paused(key, consumer, paused=False)
+                return
+            except Exception:
+                logger.warning(
+                    "failed to resume %s[%s] (attempt %s/3)",
+                    key.topic,
+                    key.partition,
+                    attempt,
+                    exc_info=True,
+                )
+                if attempt < 3:
+                    await asyncio.sleep(self._base_delay)
+        logger.error(
+            "giving up resuming %s[%s] after 3 attempts - it stays "
+            "paused until the consumer restarts",
+            key.topic,
+            key.partition,
+        )

tests/kafka/test_backoff.py

@@ -12,16 +12,18 @@ def _fake_message(
     offset: int | None,
     is_manual: bool = True,
     batch_shape: type[list] | type[tuple] | None = None,
+    key: bytes | None = b"k",
 ):
     single = SimpleNamespace(
         topic=lambda: topic,
         partition=lambda: partition,
         offset=lambda: offset,
+        key=lambda: key,
     )
     raw: (
         SimpleNamespace | list[SimpleNamespace] | tuple[SimpleNamespace, ...]
     ) = single if batch_shape is None else batch_shape([single])
-    return SimpleNamespace(raw_message=raw, is_manual=is_manual)
+    return SimpleNamespace(raw_message=raw, is_manual=is_manual, consumer=None)
 
 
 @pytest.fixture