diff --git a/docs/adr/01-kafka-retry-backoff.md b/docs/adr/01-kafka-retry-backoff.md new file mode 100644 index 0000000..7ce1ef9 --- /dev/null +++ b/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. diff --git a/docs/signals.md b/docs/signals.md index f35cd53..dd77ffd 100644 --- a/docs/signals.md +++ b/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. diff --git a/fastloom/signals/kafka/depends.py b/fastloom/signals/kafka/depends.py index 0be3b57..1cddcc2 100644 --- a/fastloom/signals/kafka/depends.py +++ b/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, + ) diff --git a/tests/kafka/test_backoff.py b/tests/kafka/test_backoff.py index 8705ceb..3cfdcca 100644 --- a/tests/kafka/test_backoff.py +++ b/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 diff --git a/tests/kafka/test_pause_resume_backoff.py b/tests/kafka/test_pause_resume_backoff.py new file mode 100644 index 0000000..dc22153 --- /dev/null +++ b/tests/kafka/test_pause_resume_backoff.py @@ -0,0 +1,326 @@ +import asyncio + +from confluent_kafka import KafkaError, KafkaException, Message +from confluent_kafka.admin import AdminClient, NewTopic +from faststream.confluent.fastapi import KafkaMessage + +from fastloom.signals.kafka.depends import KafkaSubscriber +from fastloom.signals.kafka.settings import KafkaSubscriptable + + +class _ConsumerProxy: + """Forwards to the real confluent Consumer, recording pause()/resume() + calls and optionally forcing one of them to fail - keeps the rest of + the stack (broker, thread pool, message flow) real.""" + + def __init__(self, real, *, fail_pause=False, fail_resume=False): + self._real = real + self._fail_pause = fail_pause + self._fail_resume = fail_resume + self.pause_attempts = 0 + self.resume_attempts = 0 + + def pause(self, partitions): + self.pause_attempts += 1 + if self._fail_pause: + raise RuntimeError("pause boom") + return self._real.pause(partitions) + + def resume(self, partitions): + self.resume_attempts += 1 + if self._fail_resume: + raise RuntimeError("resume boom") + return self._real.resume(partitions) + + def __getattr__(self, name): + return getattr(self._real, name) + + +def _create_topic(admin: AdminClient, topic: str, num_partitions: int) -> None: + (future,) = admin.create_topics( + [NewTopic(topic, num_partitions=num_partitions, replication_factor=1)] + ).values() + try: + future.result() + except KafkaException as e: + if e.args[0].code() != KafkaError.TOPIC_ALREADY_EXISTS: + raise + + +async def _wait_for_topic( + admin: AdminClient, topic: str, partitions: int +) -> None: + deadline = asyncio.get_event_loop().time() + 10 + while asyncio.get_event_loop().time() < deadline: + metadata = admin.list_topics(topic=topic, timeout=2).topics.get(topic) + if metadata is not None and len(metadata.partitions) >= partitions: + return + await asyncio.sleep(0.2) + raise TimeoutError(f"topic {topic} never reached {partitions} partitions") + + +async def _wait_until(predicate, timeout: float = 15) -> None: + deadline = asyncio.get_event_loop().time() + timeout + while asyncio.get_event_loop().time() < deadline: + if predicate(): + return + await asyncio.sleep(0.1) + raise TimeoutError("condition never became true") + + +async def test_non_keyed_backoff_does_not_block_other_partitions( + kafka_subscriber, kafka_container +): + topic = "cross-partition-test" + admin = AdminClient( + {"bootstrap.servers": kafka_container.get_bootstrap_server()} + ) + _create_topic(admin, topic, num_partitions=2) + await _wait_for_topic(admin, topic, partitions=2) + + router = kafka_subscriber.router + failed_event = asyncio.Event() + partition_1_done = asyncio.Event() + partition_0_recovered = asyncio.Event() + times: dict[str, float] = {} + + @router.subscriber( + topic, + group_id="cross-partition-test", + auto_offset_reset="earliest", + ) + async def handler(msg: KafkaMessage) -> None: + raw = msg.raw_message + assert isinstance(raw, Message) + if raw.partition() == 0: + if not failed_event.is_set(): + times["failed_at"] = asyncio.get_event_loop().time() + failed_event.set() + raise ValueError("boom - forces partition 0 into backoff") + partition_0_recovered.set() + if raw.partition() == 1: + times["partition_1_at"] = asyncio.get_event_loop().time() + partition_1_done.set() + + publisher = router.publisher(topic) + await router.broker.start() + try: + # no key on either publish - non-keyed, the case this backoff + # design applies to (see docs/adr/01-kafka-retry-backoff) + await publisher.publish("p0-msg", partition=0) + # only publish partition 1's message once partition 0 has + # actually failed and entered backoff - guarantees the ordering + # this test depends on instead of racing two publishes at once + await asyncio.wait_for(failed_event.wait(), timeout=15) + + await publisher.publish("p1-msg", partition=1) + await asyncio.wait_for(partition_1_done.wait(), timeout=15) + + # partition 0's original message must still get redelivered and + # succeed once the backoff's resume() fires - proves the pause + # is temporary, not a permanent stall of that partition + await asyncio.wait_for(partition_0_recovered.wait(), timeout=15) + finally: + await router.broker.stop() + + elapsed = times["partition_1_at"] - times["failed_at"] + # kafka_subscriber's default base_delay is 5s - partition 1 arriving + # well under that, right after partition 0 failed, proves the fetch + # loop kept polling instead of blocking on partition 0's backoff. + assert elapsed < 4 + + +async def test_keyed_message_never_touches_the_consumer(kafka_container): + topic = "keyed-never-touches-test" + admin = AdminClient( + {"bootstrap.servers": kafka_container.get_bootstrap_server()} + ) + _create_topic(admin, topic, num_partitions=1) + await _wait_for_topic(admin, topic, partitions=1) + + settings = KafkaSubscriptable( + ENVIRONMENT="test", + PROJECT_NAME="fastloom_test", + KAFKA_URI=kafka_container.get_bootstrap_server(), + ) + subscriber = KafkaSubscriber(settings, base_delay=1, max_delay=8) + proxy: _ConsumerProxy | None = None + failed_event = asyncio.Event() + + @subscriber.router.subscriber( + topic, + group_id="keyed-test", + auto_offset_reset="earliest", + ) + async def handler(msg: KafkaMessage) -> None: + nonlocal proxy + if proxy is None: + proxy = _ConsumerProxy(msg.consumer.consumer) + msg.consumer.consumer = proxy + if not failed_event.is_set(): + failed_event.set() + raise ValueError("boom") + + publisher = subscriber.router.publisher(topic) + await subscriber.router.broker.start() + try: + await publisher.publish("keyed-msg", key=b"some-key") + await asyncio.wait_for(failed_event.wait(), timeout=15) + # give the (keyed -> inline sleep) path a moment - long enough to + # prove it never reaches for pause/resume at all + await asyncio.sleep(1) + finally: + await subscriber.router.broker.stop() + KafkaSubscriber.unbind() + + assert proxy is not None + assert proxy.pause_attempts == 0 + assert proxy.resume_attempts == 0 + + +async def test_non_keyed_message_pauses_then_resumes(kafka_container): + topic = "non-keyed-pauses-then-resumes-test" + admin = AdminClient( + {"bootstrap.servers": kafka_container.get_bootstrap_server()} + ) + _create_topic(admin, topic, num_partitions=1) + await _wait_for_topic(admin, topic, partitions=1) + + settings = KafkaSubscriptable( + ENVIRONMENT="test", + PROJECT_NAME="fastloom_test", + KAFKA_URI=kafka_container.get_bootstrap_server(), + ) + subscriber = KafkaSubscriber(settings, base_delay=1, max_delay=8) + proxy: _ConsumerProxy | None = None + failed_event = asyncio.Event() + recovered_event = asyncio.Event() + + @subscriber.router.subscriber( + topic, + group_id="non-keyed-pause-resume-test", + auto_offset_reset="earliest", + ) + async def handler(msg: KafkaMessage) -> None: + nonlocal proxy + if proxy is None: + proxy = _ConsumerProxy(msg.consumer.consumer) + msg.consumer.consumer = proxy + if not failed_event.is_set(): + failed_event.set() + raise ValueError("boom") + recovered_event.set() + + publisher = subscriber.router.publisher(topic) + await subscriber.router.broker.start() + try: + await publisher.publish("non-keyed-msg") # no key + await asyncio.wait_for(failed_event.wait(), timeout=15) + await asyncio.wait_for(recovered_event.wait(), timeout=15) + finally: + await subscriber.router.broker.stop() + KafkaSubscriber.unbind() + + assert proxy is not None + assert proxy.pause_attempts == 1 + assert proxy.resume_attempts == 1 + + +async def test_pause_failure_falls_back_to_inline_backoff(kafka_container): + topic = "pause-failure-test" + admin = AdminClient( + {"bootstrap.servers": kafka_container.get_bootstrap_server()} + ) + _create_topic(admin, topic, num_partitions=1) + await _wait_for_topic(admin, topic, partitions=1) + + settings = KafkaSubscriptable( + ENVIRONMENT="test", + PROJECT_NAME="fastloom_test", + KAFKA_URI=kafka_container.get_bootstrap_server(), + ) + subscriber = KafkaSubscriber(settings, base_delay=1, max_delay=8) + proxy: _ConsumerProxy | None = None + failed_event = asyncio.Event() + recovered_event = asyncio.Event() + + @subscriber.router.subscriber( + topic, + group_id="pause-failure-test", + auto_offset_reset="earliest", + ) + async def handler(msg: KafkaMessage) -> None: + nonlocal proxy + if proxy is None: + proxy = _ConsumerProxy(msg.consumer.consumer, fail_pause=True) + msg.consumer.consumer = proxy + if not failed_event.is_set(): + failed_event.set() + raise ValueError("boom") + recovered_event.set() + + publisher = subscriber.router.publisher(topic) + await subscriber.router.broker.start() + try: + await publisher.publish("pause-failure-msg") # no key + await asyncio.wait_for(failed_event.wait(), timeout=15) + # pause() raised - the original exception still had to propagate + # for redelivery instead of being swallowed, so the message must + # come back and succeed via the inline-sleep fallback + await asyncio.wait_for(recovered_event.wait(), timeout=15) + finally: + await subscriber.router.broker.stop() + KafkaSubscriber.unbind() + + assert proxy is not None + assert proxy.pause_attempts == 1 + assert proxy.resume_attempts == 0 # never scheduled - pause failed first + + +async def test_resume_failure_retries_then_gives_up(kafka_container, caplog): + topic = "resume-failure-test" + admin = AdminClient( + {"bootstrap.servers": kafka_container.get_bootstrap_server()} + ) + _create_topic(admin, topic, num_partitions=1) + await _wait_for_topic(admin, topic, partitions=1) + + settings = KafkaSubscriptable( + ENVIRONMENT="test", + PROJECT_NAME="fastloom_test", + KAFKA_URI=kafka_container.get_bootstrap_server(), + ) + subscriber = KafkaSubscriber(settings, base_delay=1, max_delay=8) + proxy: _ConsumerProxy | None = None + failed_event = asyncio.Event() + + @subscriber.router.subscriber( + topic, + group_id="resume-failure-test", + auto_offset_reset="earliest", + ) + async def handler(msg: KafkaMessage) -> None: + nonlocal proxy + if proxy is None: + proxy = _ConsumerProxy(msg.consumer.consumer, fail_resume=True) + msg.consumer.consumer = proxy + if not failed_event.is_set(): + failed_event.set() + raise ValueError("boom") + + publisher = subscriber.router.publisher(topic) + await subscriber.router.broker.start() + try: + await publisher.publish("resume-failure-msg") # no key + await asyncio.wait_for(failed_event.wait(), timeout=15) + await _wait_until( + lambda: proxy is not None and proxy.resume_attempts >= 3 + ) + finally: + await subscriber.router.broker.stop() + KafkaSubscriber.unbind() + + assert proxy is not None + assert proxy.pause_attempts == 1 + assert proxy.resume_attempts == 3 # retried 3x, every attempt raised + assert "giving up resuming" in caplog.text