From 5031a5672f831a1b52c9e41071754cfc668c73b9 Mon Sep 17 00:00:00 2001 From: aradng Date: Wed, 22 Jul 2026 13:42:00 +0330 Subject: [PATCH 1/3] feat: pause Kafka partitions instead of blocking backoff for non-keyed messages Non-keyed topics have no ordering requirement, so blocking a whole consumer's poll loop while one message backs off needlessly delays unrelated messages on other partitions. For a message with no Kafka key, KafkaSubscriber now pauses just that partition (confluent_kafka's Consumer.pause/.resume) and keeps polling, instead of sleeping inline. Keyed messages are unchanged. See docs/adr/01-kafka-non-keyed-retry-backoff.md for the alternatives considered (numbered retry topics, an external delay store, max_workers concurrency) and why each was rejected. Co-Authored-By: Claude Sonnet 5 --- docs/adr/01-kafka-non-keyed-retry-backoff.md | 104 +++++++++++++++++++ docs/signals.md | 11 +- fastloom/signals/kafka/depends.py | 71 ++++++++++++- tests/kafka/test_backoff.py | 84 ++++++++++++++- tests/kafka/test_pause_resume_backoff.py | 62 +++++++++++ 5 files changed, 324 insertions(+), 8 deletions(-) create mode 100644 docs/adr/01-kafka-non-keyed-retry-backoff.md create mode 100644 tests/kafka/test_pause_resume_backoff.py diff --git a/docs/adr/01-kafka-non-keyed-retry-backoff.md b/docs/adr/01-kafka-non-keyed-retry-backoff.md new file mode 100644 index 0000000..6b989ab --- /dev/null +++ b/docs/adr/01-kafka-non-keyed-retry-backoff.md @@ -0,0 +1,104 @@ +# 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. diff --git a/docs/signals.md b/docs/signals.md index f35cd53..8be3231 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-non-keyed-retry-backoff.md](adr/01-kafka-non-keyed-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-non-keyed-retry-backoff.md](adr/01-kafka-non-keyed-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..eb69a50 100644 --- a/fastloom/signals/kafka/depends.py +++ b/fastloom/signals/kafka/depends.py @@ -19,6 +19,7 @@ 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 @@ -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) @@ -344,13 +347,17 @@ async def consume_scope(self, call_next, msg): ) @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 +373,11 @@ 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: + return cls._first_record(message).key() is not None + + async def _backoff(self, key: _MessageKey, message: KafkaMessage) -> None: partition_key = key.partition_key last = self._retry_state.get(partition_key) attempt = ( @@ -385,4 +396,56 @@ async def _backoff(self, key: _MessageKey) -> None: delay, attempt, ) + + if self._is_keyed(message): + # keyed: the partition is already one ordered stream, so + # blocking it here costs nothing extra - see docs/adr/01. + await asyncio.sleep(delay) + return + + await self._pause_partition(key, message) + self._schedule_resume(key, message, delay) + + async def _pause_partition( + self, key: _MessageKey, message: KafkaMessage + ) -> None: + from confluent_kafka import TopicPartition + + # librdkafka's client isn't thread-safe - pause/resume must run + # through the same single-thread executor FastStream already + # serializes every other consumer call on. + consumer = message.consumer + await asyncio.get_running_loop().run_in_executor( + consumer._thread_pool, # type: ignore[attr-defined] + consumer.consumer.pause, # type: ignore[attr-defined] + [TopicPartition(key.topic, key.partition)], + ) + + def _schedule_resume( + self, key: _MessageKey, message: KafkaMessage, delay: float + ) -> None: + task = asyncio.create_task(self._resume_partition(key, message, delay)) + self._pending_resumes.add(task) + task.add_done_callback(self._pending_resumes.discard) + + async def _resume_partition( + self, key: _MessageKey, message: KafkaMessage, delay: float + ) -> None: + from confluent_kafka import TopicPartition + await asyncio.sleep(delay) + consumer = message.consumer + try: + await asyncio.get_running_loop().run_in_executor( + consumer._thread_pool, # type: ignore[attr-defined] + consumer.consumer.resume, # type: ignore[attr-defined] + [TopicPartition(key.topic, key.partition)], + ) + except Exception: + logger.warning( + "failed to resume %s[%s] after backoff - consumer is " + "likely shutting down", + key.topic, + key.partition, + exc_info=True, + ) diff --git a/tests/kafka/test_backoff.py b/tests/kafka/test_backoff.py index 8705ceb..bced55e 100644 --- a/tests/kafka/test_backoff.py +++ b/tests/kafka/test_backoff.py @@ -1,27 +1,53 @@ +import asyncio +from concurrent.futures import ThreadPoolExecutor from types import SimpleNamespace import pytest +from confluent_kafka import TopicPartition from fastloom.signals.kafka.depends import KafkaSubscriber from fastloom.signals.kafka.settings import KafkaSubscriptable +class _FakeRawConsumer: + def __init__(self): + self.paused: list[list[TopicPartition]] = [] + self.resumed: list[list[TopicPartition]] = [] + + def pause(self, partitions: list[TopicPartition]) -> None: + self.paused.append(partitions) + + def resume(self, partitions: list[TopicPartition]) -> None: + self.resumed.append(partitions) + + +class _FakeConsumer: + def __init__(self): + self.consumer = _FakeRawConsumer() + self._thread_pool = ThreadPoolExecutor(max_workers=1) + + def _fake_message( topic: str | None, partition: int | None, offset: int | None, is_manual: bool = True, batch_shape: type[list] | type[tuple] | None = None, + key: bytes | None = b"k", + consumer: _FakeConsumer | None = None, ): 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=consumer + ) @pytest.fixture @@ -147,6 +173,62 @@ async def test_batch_message_backs_off_regardless_of_container_type( assert deterministic_subscriber.slept == [1] +async def _drain_resumes(subscriber): + while subscriber._pending_resumes: + await asyncio.gather(*list(subscriber._pending_resumes)) + + +async def test_keyed_message_never_touches_the_consumer( + deterministic_subscriber, +): + fake_consumer = _FakeConsumer() + + await _fail(deterministic_subscriber, key=b"k", consumer=fake_consumer) + + assert deterministic_subscriber.slept == [1] # blocked inline, as before + assert fake_consumer.consumer.paused == [] + assert fake_consumer.consumer.resumed == [] + + +async def test_non_keyed_message_pauses_instead_of_blocking( + deterministic_subscriber, +): + fake_consumer = _FakeConsumer() + + await _fail( + deterministic_subscriber, key=None, consumer=fake_consumer, offset=1 + ) + + # backoff returns immediately - no inline sleep, no resume yet either + assert deterministic_subscriber.slept == [] + assert fake_consumer.consumer.paused == [[TopicPartition("t", 0)]] + assert fake_consumer.consumer.resumed == [] + + await _drain_resumes(deterministic_subscriber) + + assert deterministic_subscriber.slept == [1] + assert fake_consumer.consumer.resumed == [[TopicPartition("t", 0)]] + + +async def test_non_keyed_backoff_still_doubles_on_repeated_failures( + deterministic_subscriber, +): + fake_consumer = _FakeConsumer() + + for _ in range(4): + await _fail( + deterministic_subscriber, + key=None, + consumer=fake_consumer, + offset=42, + ) + await _drain_resumes(deterministic_subscriber) + + assert deterministic_subscriber.slept == [1, 2, 4, 8] + assert len(fake_consumer.consumer.paused) == 4 + assert len(fake_consumer.consumer.resumed) == 4 + + async def test_enable_idempotence_forces_acks_all(): settings = KafkaSubscriptable( ENVIRONMENT="test", PROJECT_NAME="p", KAFKA_URI="localhost:1" diff --git a/tests/kafka/test_pause_resume_backoff.py b/tests/kafka/test_pause_resume_backoff.py new file mode 100644 index 0000000..273a72a --- /dev/null +++ b/tests/kafka/test_pause_resume_backoff.py @@ -0,0 +1,62 @@ +import asyncio +from typing import cast + +from confluent_kafka import Message +from confluent_kafka.admin import AdminClient, NewTopic +from faststream.confluent.fastapi import KafkaMessage + +TOPIC = "pause-resume-backoff-test" + + +async def test_non_keyed_backoff_does_not_block_other_partitions( + kafka_subscriber, kafka_container +): + admin = AdminClient( + {"bootstrap.servers": kafka_container.get_bootstrap_server()} + ) + admin.create_topics( + [NewTopic(TOPIC, num_partitions=2, replication_factor=1)] + ) + await asyncio.sleep(2) # topic metadata propagation + + router = kafka_subscriber.router + failed_event = asyncio.Event() + partition_1_done = asyncio.Event() + times: dict[str, float] = {} + + @router.subscriber( + TOPIC, + group_id="pause-resume-backoff-test", + auto_offset_reset="earliest", + ) + async def handler(msg: KafkaMessage) -> None: + raw = cast(Message, msg.raw_message) + if raw.partition() == 0 and not failed_event.is_set(): + times["failed_at"] = asyncio.get_event_loop().time() + failed_event.set() + raise ValueError("boom - forces partition 0 into backoff") + 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-non-keyed-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) + 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 From 8008806149f5304da655dd23e54c6e5bbeadf558 Mon Sep 17 00:00:00 2001 From: aradng Date: Wed, 22 Jul 2026 17:42:21 +0330 Subject: [PATCH 2/3] fix: harden pause/resume backoff error paths per review - pause() failure no longer swallows the original handler exception - falls back to inline backoff instead, preserving the NACK-and-redeliver contract. - resume() retries up to 3 times with backoff before giving up, instead of silently stranding a partition paused forever on the first failure. - KafkaSubscriber now cancels pending resume tasks when the broker stops, instead of leaving them to fire against an already-closed consumer. - _is_keyed now checks every record in a batch (any() defaults to the safe/blocking choice) instead of classifying a mixed batch off record 0. - collapsed _pause_partition/_resume_partition's duplicated TopicPartition-build-and-run_in_executor shape into one _set_partition_paused helper, reusing FastStream's own run_in_executor instead of hand-rolling it. - tests: fixed-consumer thread pools now get shut down, added coverage for both new failure paths, and the integration test now asserts partition 0's original message actually gets redelivered and succeeds after resume (previously only checked partition 1 wasn't blocked), plus polls for topic readiness instead of a fixed sleep. Co-Authored-By: Claude Sonnet 5 --- fastloom/signals/kafka/depends.py | 103 ++++++++++++++++------- tests/kafka/test_backoff.py | 62 +++++++++++--- tests/kafka/test_pause_resume_backoff.py | 49 ++++++++--- 3 files changed, 161 insertions(+), 53 deletions(-) diff --git a/fastloom/signals/kafka/depends.py b/fastloom/signals/kafka/depends.py index eb69a50..cb30599 100644 --- a/fastloom/signals/kafka/depends.py +++ b/fastloom/signals/kafka/depends.py @@ -22,7 +22,7 @@ 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, @@ -346,6 +346,15 @@ 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 _first_record(message: KafkaMessage) -> Message: raw = message.raw_message @@ -375,7 +384,13 @@ def _clear_retry_state(self, key: _MessageKey) -> None: @classmethod def _is_keyed(cls, message: KafkaMessage) -> bool: - return cls._first_record(message).key() is not None + raw = message.raw_message + # a batch mixing keyed and non-keyed records has to pick one + # answer for the whole batch (Kafka acks/commits it as one unit) + # - any() defaults to the safe (blocking) choice instead of + # picking arbitrarily off the first record. + 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 @@ -403,49 +418,77 @@ async def _backoff(self, key: _MessageKey, message: KafkaMessage) -> None: await asyncio.sleep(delay) return - await self._pause_partition(key, message) - self._schedule_resume(key, message, delay) + consumer = message.consumer + try: + await self._set_partition_paused(key, consumer, paused=True) + except Exception: + # pause() itself failing must not propagate here - it would + # replace the caller's original exception in consume_scope's + # except block instead of letting it re-raise for redelivery. + 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 _pause_partition( - self, key: _MessageKey, message: KafkaMessage + 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 # librdkafka's client isn't thread-safe - pause/resume must run # through the same single-thread executor FastStream already - # serializes every other consumer call on. - consumer = message.consumer - await asyncio.get_running_loop().run_in_executor( + # serializes every other consumer call on. ConsumerProtocol only + # declares commit()/seek() - this reaches past it to the concrete + # AsyncConfluentConsumer FastStream actually constructs. + 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] - consumer.consumer.pause, # type: ignore[attr-defined] + op, [TopicPartition(key.topic, key.partition)], ) def _schedule_resume( - self, key: _MessageKey, message: KafkaMessage, delay: float + self, key: _MessageKey, consumer: ConsumerProtocol, delay: float ) -> None: - task = asyncio.create_task(self._resume_partition(key, message, delay)) + 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, message: KafkaMessage, delay: float + self, key: _MessageKey, consumer: ConsumerProtocol, delay: float ) -> None: - from confluent_kafka import TopicPartition - await asyncio.sleep(delay) - consumer = message.consumer - try: - await asyncio.get_running_loop().run_in_executor( - consumer._thread_pool, # type: ignore[attr-defined] - consumer.consumer.resume, # type: ignore[attr-defined] - [TopicPartition(key.topic, key.partition)], - ) - except Exception: - logger.warning( - "failed to resume %s[%s] after backoff - consumer is " - "likely shutting down", - key.topic, - key.partition, - exc_info=True, - ) + 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 bced55e..c615641 100644 --- a/tests/kafka/test_backoff.py +++ b/tests/kafka/test_backoff.py @@ -10,23 +10,36 @@ class _FakeRawConsumer: - def __init__(self): + def __init__(self, fail_pause: bool = False, fail_resume: bool = False): self.paused: list[list[TopicPartition]] = [] self.resumed: list[list[TopicPartition]] = [] + self._fail_pause = fail_pause + self._fail_resume = fail_resume def pause(self, partitions: list[TopicPartition]) -> None: + if self._fail_pause: + raise RuntimeError("pause boom") self.paused.append(partitions) def resume(self, partitions: list[TopicPartition]) -> None: + if self._fail_resume: + raise RuntimeError("resume boom") self.resumed.append(partitions) class _FakeConsumer: - def __init__(self): - self.consumer = _FakeRawConsumer() + def __init__(self, fail_pause: bool = False, fail_resume: bool = False): + self.consumer = _FakeRawConsumer(fail_pause, fail_resume) self._thread_pool = ThreadPoolExecutor(max_workers=1) +@pytest.fixture +def fake_consumer(): + consumer = _FakeConsumer() + yield consumer + consumer._thread_pool.shutdown() + + def _fake_message( topic: str | None, partition: int | None, @@ -179,10 +192,8 @@ async def _drain_resumes(subscriber): async def test_keyed_message_never_touches_the_consumer( - deterministic_subscriber, + deterministic_subscriber, fake_consumer ): - fake_consumer = _FakeConsumer() - await _fail(deterministic_subscriber, key=b"k", consumer=fake_consumer) assert deterministic_subscriber.slept == [1] # blocked inline, as before @@ -191,10 +202,8 @@ async def test_keyed_message_never_touches_the_consumer( async def test_non_keyed_message_pauses_instead_of_blocking( - deterministic_subscriber, + deterministic_subscriber, fake_consumer ): - fake_consumer = _FakeConsumer() - await _fail( deterministic_subscriber, key=None, consumer=fake_consumer, offset=1 ) @@ -211,10 +220,8 @@ async def test_non_keyed_message_pauses_instead_of_blocking( async def test_non_keyed_backoff_still_doubles_on_repeated_failures( - deterministic_subscriber, + deterministic_subscriber, fake_consumer ): - fake_consumer = _FakeConsumer() - for _ in range(4): await _fail( deterministic_subscriber, @@ -229,6 +236,37 @@ async def test_non_keyed_backoff_still_doubles_on_repeated_failures( assert len(fake_consumer.consumer.resumed) == 4 +async def test_pause_failure_falls_back_to_inline_backoff( + deterministic_subscriber, +): + fake_consumer = _FakeConsumer(fail_pause=True) + + await _fail(deterministic_subscriber, key=None, consumer=fake_consumer) + + # pause() blew up - the original ValueError still had to propagate, + # and backoff must fall back to blocking instead of losing the retry + assert deterministic_subscriber.slept == [1] + assert fake_consumer.consumer.paused == [] + assert deterministic_subscriber._pending_resumes == set() + + fake_consumer._thread_pool.shutdown() + + +async def test_resume_failure_retries_then_gives_up(deterministic_subscriber): + fake_consumer = _FakeConsumer(fail_resume=True) + + await _fail(deterministic_subscriber, key=None, consumer=fake_consumer) + await _drain_resumes(deterministic_subscriber) + + # 1 initial backoff sleep + 2 retry sleeps between the 3 failed + # resume attempts (base_delay=1 for every retry, not exponential) + assert deterministic_subscriber.slept == [1, 1, 1] + assert fake_consumer.consumer.paused == [[TopicPartition("t", 0)]] + assert fake_consumer.consumer.resumed == [] # every attempt raised + + fake_consumer._thread_pool.shutdown() + + async def test_enable_idempotence_forces_acks_all(): settings = KafkaSubscriptable( ENVIRONMENT="test", PROJECT_NAME="p", KAFKA_URI="localhost:1" diff --git a/tests/kafka/test_pause_resume_backoff.py b/tests/kafka/test_pause_resume_backoff.py index 273a72a..f885f1d 100644 --- a/tests/kafka/test_pause_resume_backoff.py +++ b/tests/kafka/test_pause_resume_backoff.py @@ -1,27 +1,46 @@ import asyncio -from typing import cast -from confluent_kafka import Message +from confluent_kafka import KafkaError, KafkaException, Message from confluent_kafka.admin import AdminClient, NewTopic from faststream.confluent.fastapi import KafkaMessage TOPIC = "pause-resume-backoff-test" +def _create_topic(admin: AdminClient) -> None: + (future,) = admin.create_topics( + [NewTopic(TOPIC, num_partitions=2, 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, 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 test_non_keyed_backoff_does_not_block_other_partitions( kafka_subscriber, kafka_container ): admin = AdminClient( {"bootstrap.servers": kafka_container.get_bootstrap_server()} ) - admin.create_topics( - [NewTopic(TOPIC, num_partitions=2, replication_factor=1)] - ) - await asyncio.sleep(2) # topic metadata propagation + _create_topic(admin) + await _wait_for_topic(admin, 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( @@ -30,11 +49,14 @@ async def test_non_keyed_backoff_does_not_block_other_partitions( auto_offset_reset="earliest", ) async def handler(msg: KafkaMessage) -> None: - raw = cast(Message, msg.raw_message) - if raw.partition() == 0 and not failed_event.is_set(): - times["failed_at"] = asyncio.get_event_loop().time() - failed_event.set() - raise ValueError("boom - forces partition 0 into backoff") + 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() @@ -52,6 +74,11 @@ async def handler(msg: KafkaMessage) -> None: 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() From cf8b3f7e5bd7571d22547ae177ac7edd14704cf2 Mon Sep 17 00:00:00 2001 From: aradng Date: Wed, 22 Jul 2026 18:06:03 +0330 Subject: [PATCH 3/3] refactor: pause/resume tests against real broker, trim inline comments - moved keyed/non-keyed pause-resume test coverage from SimpleNamespace fakes into real-broker testcontainer tests (test_pause_resume_backoff.py) - a thin proxy wraps the real confluent Consumer to record/force-fail pause()/resume(), everything else (broker, thread pool, message flow) stays real. Each test uses its own topic to avoid cross-test message replay from auto_offset_reset="earliest" on a shared topic. - test_backoff.py reverts to its original fake-message shape - it only ever exercises the keyed (inline-sleep) branch now, so none of the consumer/thread-pool machinery is needed there anymore. - trimmed the new inline comments in depends.py to one-line pointers at docs/adr/01-kafka-retry-backoff.md instead of re-explaining the reasoning at each call site. - renamed the ADR to docs/adr/01-kafka-retry-backoff.md. Co-Authored-By: Claude Sonnet 5 --- ...y-backoff.md => 01-kafka-retry-backoff.md} | 14 + docs/signals.md | 4 +- fastloom/signals/kafka/depends.py | 19 +- tests/kafka/test_backoff.py | 120 +------- tests/kafka/test_pause_resume_backoff.py | 263 +++++++++++++++++- 5 files changed, 272 insertions(+), 148 deletions(-) rename docs/adr/{01-kafka-non-keyed-retry-backoff.md => 01-kafka-retry-backoff.md} (85%) diff --git a/docs/adr/01-kafka-non-keyed-retry-backoff.md b/docs/adr/01-kafka-retry-backoff.md similarity index 85% rename from docs/adr/01-kafka-non-keyed-retry-backoff.md rename to docs/adr/01-kafka-retry-backoff.md index 6b989ab..7ce1ef9 100644 --- a/docs/adr/01-kafka-non-keyed-retry-backoff.md +++ b/docs/adr/01-kafka-retry-backoff.md @@ -102,3 +102,17 @@ assigned the same partition reads normally right away, no stuck state. 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 8be3231..dd77ffd 100644 --- a/docs/signals.md +++ b/docs/signals.md @@ -168,7 +168,7 @@ How the backoff actually waits depends on whether the message carries a Kafka ke 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. -- See [docs/adr/01-kafka-non-keyed-retry-backoff.md](adr/01-kafka-non-keyed-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. +- 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). @@ -201,4 +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-non-keyed-retry-backoff.md](adr/01-kafka-non-keyed-retry-backoff.md) — why Kafka backoff branches on keyed vs. non-keyed messages. +- [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 cb30599..1cddcc2 100644 --- a/fastloom/signals/kafka/depends.py +++ b/fastloom/signals/kafka/depends.py @@ -385,10 +385,7 @@ def _clear_retry_state(self, key: _MessageKey) -> None: @classmethod def _is_keyed(cls, message: KafkaMessage) -> bool: raw = message.raw_message - # a batch mixing keyed and non-keyed records has to pick one - # answer for the whole batch (Kafka acks/commits it as one unit) - # - any() defaults to the safe (blocking) choice instead of - # picking arbitrarily off the first record. + # 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) @@ -413,8 +410,7 @@ async def _backoff(self, key: _MessageKey, message: KafkaMessage) -> None: ) if self._is_keyed(message): - # keyed: the partition is already one ordered stream, so - # blocking it here costs nothing extra - see docs/adr/01. + # keyed: blocking costs nothing extra - see docs/adr/01 await asyncio.sleep(delay) return @@ -422,9 +418,8 @@ async def _backoff(self, key: _MessageKey, message: KafkaMessage) -> None: try: await self._set_partition_paused(key, consumer, paused=True) except Exception: - # pause() itself failing must not propagate here - it would - # replace the caller's original exception in consume_scope's - # except block instead of letting it re-raise for redelivery. + # 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, @@ -446,11 +441,7 @@ async def _set_partition_paused( from confluent_kafka import TopicPartition from faststream._internal.utils.functions import run_in_executor - # librdkafka's client isn't thread-safe - pause/resume must run - # through the same single-thread executor FastStream already - # serializes every other consumer call on. ConsumerProtocol only - # declares commit()/seek() - this reaches past it to the concrete - # AsyncConfluentConsumer FastStream actually constructs. + # 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( diff --git a/tests/kafka/test_backoff.py b/tests/kafka/test_backoff.py index c615641..3cfdcca 100644 --- a/tests/kafka/test_backoff.py +++ b/tests/kafka/test_backoff.py @@ -1,45 +1,11 @@ -import asyncio -from concurrent.futures import ThreadPoolExecutor from types import SimpleNamespace import pytest -from confluent_kafka import TopicPartition from fastloom.signals.kafka.depends import KafkaSubscriber from fastloom.signals.kafka.settings import KafkaSubscriptable -class _FakeRawConsumer: - def __init__(self, fail_pause: bool = False, fail_resume: bool = False): - self.paused: list[list[TopicPartition]] = [] - self.resumed: list[list[TopicPartition]] = [] - self._fail_pause = fail_pause - self._fail_resume = fail_resume - - def pause(self, partitions: list[TopicPartition]) -> None: - if self._fail_pause: - raise RuntimeError("pause boom") - self.paused.append(partitions) - - def resume(self, partitions: list[TopicPartition]) -> None: - if self._fail_resume: - raise RuntimeError("resume boom") - self.resumed.append(partitions) - - -class _FakeConsumer: - def __init__(self, fail_pause: bool = False, fail_resume: bool = False): - self.consumer = _FakeRawConsumer(fail_pause, fail_resume) - self._thread_pool = ThreadPoolExecutor(max_workers=1) - - -@pytest.fixture -def fake_consumer(): - consumer = _FakeConsumer() - yield consumer - consumer._thread_pool.shutdown() - - def _fake_message( topic: str | None, partition: int | None, @@ -47,7 +13,6 @@ def _fake_message( is_manual: bool = True, batch_shape: type[list] | type[tuple] | None = None, key: bytes | None = b"k", - consumer: _FakeConsumer | None = None, ): single = SimpleNamespace( topic=lambda: topic, @@ -58,9 +23,7 @@ def _fake_message( 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, consumer=consumer - ) + return SimpleNamespace(raw_message=raw, is_manual=is_manual, consumer=None) @pytest.fixture @@ -186,87 +149,6 @@ async def test_batch_message_backs_off_regardless_of_container_type( assert deterministic_subscriber.slept == [1] -async def _drain_resumes(subscriber): - while subscriber._pending_resumes: - await asyncio.gather(*list(subscriber._pending_resumes)) - - -async def test_keyed_message_never_touches_the_consumer( - deterministic_subscriber, fake_consumer -): - await _fail(deterministic_subscriber, key=b"k", consumer=fake_consumer) - - assert deterministic_subscriber.slept == [1] # blocked inline, as before - assert fake_consumer.consumer.paused == [] - assert fake_consumer.consumer.resumed == [] - - -async def test_non_keyed_message_pauses_instead_of_blocking( - deterministic_subscriber, fake_consumer -): - await _fail( - deterministic_subscriber, key=None, consumer=fake_consumer, offset=1 - ) - - # backoff returns immediately - no inline sleep, no resume yet either - assert deterministic_subscriber.slept == [] - assert fake_consumer.consumer.paused == [[TopicPartition("t", 0)]] - assert fake_consumer.consumer.resumed == [] - - await _drain_resumes(deterministic_subscriber) - - assert deterministic_subscriber.slept == [1] - assert fake_consumer.consumer.resumed == [[TopicPartition("t", 0)]] - - -async def test_non_keyed_backoff_still_doubles_on_repeated_failures( - deterministic_subscriber, fake_consumer -): - for _ in range(4): - await _fail( - deterministic_subscriber, - key=None, - consumer=fake_consumer, - offset=42, - ) - await _drain_resumes(deterministic_subscriber) - - assert deterministic_subscriber.slept == [1, 2, 4, 8] - assert len(fake_consumer.consumer.paused) == 4 - assert len(fake_consumer.consumer.resumed) == 4 - - -async def test_pause_failure_falls_back_to_inline_backoff( - deterministic_subscriber, -): - fake_consumer = _FakeConsumer(fail_pause=True) - - await _fail(deterministic_subscriber, key=None, consumer=fake_consumer) - - # pause() blew up - the original ValueError still had to propagate, - # and backoff must fall back to blocking instead of losing the retry - assert deterministic_subscriber.slept == [1] - assert fake_consumer.consumer.paused == [] - assert deterministic_subscriber._pending_resumes == set() - - fake_consumer._thread_pool.shutdown() - - -async def test_resume_failure_retries_then_gives_up(deterministic_subscriber): - fake_consumer = _FakeConsumer(fail_resume=True) - - await _fail(deterministic_subscriber, key=None, consumer=fake_consumer) - await _drain_resumes(deterministic_subscriber) - - # 1 initial backoff sleep + 2 retry sleeps between the 3 failed - # resume attempts (base_delay=1 for every retry, not exponential) - assert deterministic_subscriber.slept == [1, 1, 1] - assert fake_consumer.consumer.paused == [[TopicPartition("t", 0)]] - assert fake_consumer.consumer.resumed == [] # every attempt raised - - fake_consumer._thread_pool.shutdown() - - async def test_enable_idempotence_forces_acks_all(): settings = KafkaSubscriptable( ENVIRONMENT="test", PROJECT_NAME="p", KAFKA_URI="localhost:1" diff --git a/tests/kafka/test_pause_resume_backoff.py b/tests/kafka/test_pause_resume_backoff.py index f885f1d..dc22153 100644 --- a/tests/kafka/test_pause_resume_backoff.py +++ b/tests/kafka/test_pause_resume_backoff.py @@ -4,12 +4,41 @@ from confluent_kafka.admin import AdminClient, NewTopic from faststream.confluent.fastapi import KafkaMessage -TOPIC = "pause-resume-backoff-test" +from fastloom.signals.kafka.depends import KafkaSubscriber +from fastloom.signals.kafka.settings import KafkaSubscriptable -def _create_topic(admin: AdminClient) -> None: +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=2, replication_factor=1)] + [NewTopic(topic, num_partitions=num_partitions, replication_factor=1)] ).values() try: future.result() @@ -18,24 +47,36 @@ def _create_topic(admin: AdminClient) -> None: raise -async def _wait_for_topic(admin: AdminClient, partitions: int) -> None: +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: + 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") + 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) - await _wait_for_topic(admin, partitions=2) + _create_topic(admin, topic, num_partitions=2) + await _wait_for_topic(admin, topic, partitions=2) router = kafka_subscriber.router failed_event = asyncio.Event() @@ -44,8 +85,8 @@ async def test_non_keyed_backoff_does_not_block_other_partitions( times: dict[str, float] = {} @router.subscriber( - TOPIC, - group_id="pause-resume-backoff-test", + topic, + group_id="cross-partition-test", auto_offset_reset="earliest", ) async def handler(msg: KafkaMessage) -> None: @@ -61,11 +102,11 @@ async def handler(msg: KafkaMessage) -> None: times["partition_1_at"] = asyncio.get_event_loop().time() partition_1_done.set() - publisher = router.publisher(TOPIC) + 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-non-keyed-retry-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 @@ -87,3 +128,199 @@ async def handler(msg: KafkaMessage) -> None: # 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