From 3c81487a94c4c24f61f14f9041bb09f29d40839c Mon Sep 17 00:00:00 2001 From: Ryan Date: Fri, 3 Jul 2026 18:20:09 -0700 Subject: [PATCH 1/3] fix: reader survives raising apply-observer callbacks and transient catch-up unassignment MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Removes two independent permanent-death paths in KafkaTable._run that are not broker disconnects (disconnects are retried and survived — verified empirically): - A raising on_set/on_delete apply-observer callback is now logged (with the record key + traceback) and counted via ViewStats.callback_errors instead of killing the reader forever; the base table still applies the record. Reader liveness is the table's contract; observer integrity stays the observer's (compute-then-commit, as GroupedKafkaTable's own hooks already do). - The catch-up gate's position() read now tolerates a transient IllegalStateError from a metadata-blip assignment shrink under group_id=None, mirroring the barrier sweep's existing guard. A self-healing blip defers the gate rather than latching the reader as failed; a non-IllegalStateError still propagates, and a permanent unassignment degrades (via catchup_timeout) rather than failing. Also syncs uv.lock's ktables version to 1.1.1 (stale since the 1.1.1 release). 100% branch coverage retained. Refs #33. --- docs/API.md | 2 +- ktables/grouped_table.py | 4 + ktables/kafka_table.py | 66 +++++++++++++--- tests/test_grouped_table.py | 3 + tests/test_kafka_table.py | 152 +++++++++++++++++++++++++++++++++--- uv.lock | 2 +- 6 files changed, 207 insertions(+), 22 deletions(-) diff --git a/docs/API.md b/docs/API.md index 46e8dd5..d516100 100644 --- a/docs/API.md +++ b/docs/API.md @@ -75,7 +75,7 @@ The reader and writer fix the key byte-layer at UTF-8 and expose only |---|---| | `ensure_topic(bootstrap_servers, topic, *, num_partitions=1, replication_factor=1, topic_configs=None, on_policy_mismatch="warn") -> EnsureTopicResult` | Idempotent explicit create, plus the policy check/reconcile below. Defaults are dev-grade — production registries want RF≥3, `min.insync.replicas=2`, `acks=all`. | | `DEFAULT_TOPIC_CONFIGS` | `{"cleanup.policy": "compact"}` (read-only mapping). | -| `ViewStats` | Frozen counters: `records_applied`, `tombstones_applied`, `keyless_records`, `key_decode_errors`, `value_decode_errors`, `catch_up_seconds`, `replayed_at_catch_up`. | +| `ViewStats` | Frozen counters: `records_applied`, `tombstones_applied`, `keyless_records`, `key_decode_errors`, `value_decode_errors`, `callback_errors`, `catch_up_seconds`, `replayed_at_catch_up`. | | `SupportsJsonModel` | Protocol the `.json()` presets require (`model_dump_json` / `model_validate_json`). | | `TableStatus` | The `status` literal type. | diff --git a/ktables/grouped_table.py b/ktables/grouped_table.py index f5ed52d..b08e5a7 100644 --- a/ktables/grouped_table.py +++ b/ktables/grouped_table.py @@ -180,6 +180,10 @@ def __repr__(self) -> str: return f"" # -- index maintenance (the on_set/on_delete handlers) -------------------- + # These are compute-then-commit: the only fallible step (codec decode) is + # caught in _decode_or_count, then the index mutation is pure dict work that + # cannot raise. So they never trip the base table's callback_errors backstop + # (KafkaTable._notify_observer) — that backstop exists for EXTERNAL observers. def _decode_or_count(self, key: str) -> tuple[str, str] | None: """Decode the composite key; count + skip foreign keys (``decode`` returns ``None``, diff --git a/ktables/kafka_table.py b/ktables/kafka_table.py index bbde99b..24f610e 100644 --- a/ktables/kafka_table.py +++ b/ktables/kafka_table.py @@ -451,13 +451,19 @@ def _utf8_encode(s: str) -> bytes: @dataclass(frozen=True, slots=True) class ViewStats: - """An immutable point-in-time snapshot of the reader's counters.""" + """An immutable point-in-time snapshot of the reader's counters. + + ``callback_errors`` counts times an ``on_set``/``on_delete`` apply-observer + raised: the base table still applied the record, but the observer skipped it + (a raise is logged and counted, never fatal — see :meth:`KafkaTable._apply`). + """ records_applied: int = 0 tombstones_applied: int = 0 keyless_records: int = 0 key_decode_errors: int = 0 value_decode_errors: int = 0 + callback_errors: int = 0 catch_up_seconds: float | None = None replayed_at_catch_up: int = 0 @@ -473,6 +479,7 @@ def __init__(self) -> None: self.keyless_records = 0 self.key_decode_errors = 0 self.value_decode_errors = 0 + self.callback_errors = 0 self.catch_up_seconds: float | None = None self.replayed_at_catch_up = 0 @@ -534,8 +541,10 @@ def __init__( self._key_decoder = key_decoder # Optional apply observers for derived views (e.g. GroupedKafkaTable): # on_set(key, value) fires on an applied value record, on_delete(key) on - # a tombstone. Synchronous, fired inside _apply after the dict mutation; - # must not raise (a raise kills the reader). See grouped_table.py. + # a tombstone. Synchronous, fired inside _apply after the dict mutation. + # A raise is logged and counted (stats.callback_errors), never fatal — the + # base record still applies but the observer misses that update, so keep + # observers compute-then-commit to stay consistent. See grouped_table.py. self._on_set = on_set self._on_delete = on_delete self._catchup_timeout = catchup_timeout @@ -908,7 +917,7 @@ def _apply(self, record: ConsumerRecord) -> None: self._data.pop(key, None) self._live.tombstones_applied += 1 if self._on_delete is not None: - self._on_delete(key) + self._notify_observer(self._on_delete, key, where=where) return try: value = self._value_decoder(record.value) @@ -922,7 +931,30 @@ def _apply(self, record: ConsumerRecord) -> None: self._data[key] = value self._live.records_applied += 1 if self._on_set is not None: - self._on_set(key, value) + self._notify_observer(self._on_set, key, value, where=where) + + def _notify_observer(self, callback: Callable[..., None], *args: object, where: str) -> None: + """Fire an apply-observer hook; a raise is logged + counted, never fatal. + + The base dict was already mutated by the caller, so the record is applied + regardless of the observer's fate. Catches ``Exception`` only — a + ``BaseException`` (cancellation/interrupt) still propagates, matching the + decoder guards above. Reader liveness is the table's contract here; the + observer's own integrity is its responsibility (compute-then-commit, as + derived views like ``GroupedKafkaTable`` do). + """ + try: + callback(*args) + except Exception: + self._live.callback_errors += 1 + # args[0] is the key on both paths (on_set(key, value) / on_delete(key)); + # naming it tells operators which key the observer missed (LWW self-heals + # it on that key's next write) — mirrors the value-decode log's key context. + logger.exception( + "apply-observer callback raised; record applied, observer skipped it (key=%s, %s)", + args[0], + where, + ) async def _run( self, @@ -931,8 +963,11 @@ async def _run( end_offsets: dict[TopicPartition, int], started: float, ) -> None: - # Escaping exceptions are captured by _on_reader_done (status - # "failed"); transient outages don't raise — getmany returns empty. + # Escaping exceptions are captured by _on_reader_done (status "failed"); + # transient outages don't raise — getmany returns empty. The two in-loop + # position() reads (catch-up gate below, barrier sweep further down) both + # tolerate a transient IllegalStateError from a metadata-blip assignment + # shrink, so a self-healing blip never latches the reader as failed. while True: batches = await consumer.getmany(timeout_ms=self._poll_timeout_ms) for records in batches.values(): @@ -948,8 +983,21 @@ async def _run( if new: end_offsets.update(await consumer.end_offsets(new)) tps = sorted(current, key=lambda tp: tp.partition) - positions = [await consumer.position(tp) for tp in tps] - if all(pos >= end_offsets[tp] for pos, tp in zip(positions, tps, strict=True)): + # A metadata blip can momentarily empty the assignment under + # group_id=None, making position() raise IllegalStateError. The + # barrier sweep below guards the same blip; guard it here too so a + # self-healing blip mid-catch-up defers the gate rather than killing + # the reader for good. Defer (positions=None) => never latch on a + # blip: the gate can only latch on a complete successful read. + # (A NON-IllegalStateError still propagates — a real fault must die, + # not spin.) A permanent unassignment (e.g. topic deleted) defers + # forever and reaches catchup_timeout => degraded, not failed — + # same trade-off barrier() already makes. + try: + positions: list[int] | None = [await consumer.position(tp) for tp in tps] + except IllegalStateError: + positions = None + if positions is not None and all(pos >= end_offsets[tp] for pos, tp in zip(positions, tps, strict=True)): self._live.catch_up_seconds = time.perf_counter() - started self._live.replayed_at_catch_up = self._live.records_applied + self._live.tombstones_applied self._caught_up.set() diff --git a/tests/test_grouped_table.py b/tests/test_grouped_table.py index 2e9f108..ce0e117 100644 --- a/tests/test_grouped_table.py +++ b/tests/test_grouped_table.py @@ -487,6 +487,9 @@ async def test_reads_reflect_seeded_members_after_barrier(self, topic, grouped_t assert await table.barrier() assert await table.wait_until_caught_up() assert table.status == "caught_up" + # Grouped's own compute-then-commit hooks never raise, so the base + # table's callback_errors backstop stays untouched on well-formed input. + assert table.stats.callback_errors == 0 assert table.get_member("billing", "a") == Endpoint(url="http://a") assert table.get_member("billing", "z") is None assert table.has_member("billing", "b") is True diff --git a/tests/test_kafka_table.py b/tests/test_kafka_table.py index e1645c7..df829dd 100644 --- a/tests/test_kafka_table.py +++ b/tests/test_kafka_table.py @@ -267,6 +267,8 @@ def _fake_consumer( catch_up: bool = True, getmany_error: BaseException | None = None, stop_error: BaseException | None = None, + position_errors: int = 0, + position_error: BaseException | None = None, ) -> type: """A configurable stand-in AIOKafkaConsumer for broker-free lifecycle tests. @@ -286,11 +288,17 @@ def _fake_consumer( reader death. Combine with a non-zero ``end_offset`` to die *during* catch-up. - ``stop_error``: if set, ``stop()`` raises it (after recording the attempt) — to prove a teardown failure never masks the caller's original error. + - ``position_errors``: number of initial ``position()`` calls that raise + (simulating the transient assignment-shrink blip mid-catch-up). Defaults to + raising ``IllegalStateError``; override the type with ``position_error`` + (e.g. a ``KafkaError`` to prove the catch-up guard does NOT swallow it). + The live ``position_calls`` counter lets a test observe the blip window. """ class _FakeConsumer: def __init__(self, *args: object, **kwargs: object) -> None: self.stopped = False + self.position_calls = 0 topic = args[0] if args else kwargs.get("topic", "t") self._tps = {TopicPartition(str(topic), p) for p in partitions} self._fetch_max_wait_ms = kwargs.get("fetch_max_wait_ms") @@ -319,6 +327,9 @@ async def getmany(self, timeout_ms: int | None = None) -> dict: async def position(self, tp: object) -> int: # Caught up once position reaches the end offset; catch_up=False pins # it below so the reader loops indefinitely (degraded/cancellation). + self.position_calls += 1 + if self.position_calls <= position_errors: + raise position_error if position_error is not None else IllegalStateError(f"Partition {tp} is not assigned") return end_offset if catch_up else 0 async def stop(self) -> None: @@ -1801,7 +1812,8 @@ def bad_key(b: bytes) -> str: class TestApplyHooks: """on_set/on_delete fire on _apply's two branches, never on a rejected - record; _apply does not wrap a raising hook (fail-loud). All broker-free.""" + record; a raising hook is swallowed + counted (reader-liveness backstop for + external observers — see spec §4.1). All broker-free.""" @staticmethod def _bytes_table(**kw: object) -> KafkaTable[bytes]: @@ -1868,15 +1880,59 @@ def bad_value(b: bytes) -> object: assert sets == [] assert table.stats.value_decode_errors == 1 - def test_apply_does_not_wrap_a_raising_hook(self) -> None: - # The deliberate fail-loud decision: _apply must NOT swallow a hook - # exception; a genuine observer bug surfaces (and kills the reader). + def test_apply_survives_raising_on_set_hook(self) -> None: + # A raising on_set must NOT kill the reader: the base record is still + # applied, the exception is swallowed and counted. Backstop for external + # observers; derived views stay compute-then-commit. See spec §4.1. def boom(k: str, v: bytes) -> None: raise RuntimeError("observer bug") table = self._bytes_table(on_set=boom) - with pytest.raises(RuntimeError, match="observer bug"): + table._apply(_fake_record(key=b"k", value=b"v")) # must NOT raise # type: ignore[arg-type] + assert table._data["k"] == b"v" # base record applied despite the hook + assert table.stats.callback_errors == 1 + + def test_apply_survives_raising_on_delete_hook(self) -> None: + def boom(k: str) -> None: + raise RuntimeError("observer bug") + + table = self._bytes_table(on_delete=boom) + table._data["k"] = b"v" # seed so the tombstone has something to remove + table._apply(_fake_record(key=b"k", value=None)) # must NOT raise # type: ignore[arg-type] + assert "k" not in table._data # tombstone applied despite the hook + assert table.stats.callback_errors == 1 + + def test_callback_errors_counter_accumulates(self) -> None: + def boom(k: str, v: bytes) -> None: + raise RuntimeError("observer bug") + + table = self._bytes_table(on_set=boom) + for i in range(3): + table._apply(_fake_record(key=b"k", value=b"v", offset=i)) # type: ignore[arg-type] + assert table.stats.callback_errors == 3 + + def test_callback_error_is_logged_with_key_and_traceback(self, caplog: pytest.LogCaptureFixture) -> None: + def boom(k: str, v: bytes) -> None: + raise RuntimeError("observer bug") + + table = self._bytes_table(on_set=boom) + with caplog.at_level(logging.ERROR): + table._apply(_fake_record(key=b"billing", value=b"v")) # type: ignore[arg-type] + assert "callback" in caplog.text # loud, human-readable + assert "billing" in caplog.text # names the key the observer missed (aids self-heal triage) + assert any(r.exc_info for r in caplog.records) # traceback attached + + def test_apply_propagates_baseexception_from_hook(self) -> None: + # Boundary: only Exception is swallowed. A BaseException (cancellation, + # KeyboardInterrupt) must still propagate — mirrors the decoder guards. + def boom(k: str, v: bytes) -> None: + raise KeyboardInterrupt("interrupt") + + table = self._bytes_table(on_set=boom) + with pytest.raises(KeyboardInterrupt): table._apply(_fake_record(key=b"k", value=b"v")) # type: ignore[arg-type] + assert table._data["k"] == b"v" # base applied before the hook ran + assert table.stats.callback_errors == 0 # BaseException is not a counted callback error # --------------------------------------------------------------------------- @@ -1956,6 +2012,74 @@ async def test_sweep_drops_cancelled_and_already_done_barriers(self) -> None: await _run_reader_until(table, consumer, [p0], {p0: 0}, lambda: table._barriers == []) assert cancelled.cancelled() and done.done() + async def test_catchup_survives_transient_position_unavailability(self) -> None: + # A metadata blip momentarily empties the assignment under group_id=None, + # so position() raises IllegalStateError mid-catch-up. The gate must defer + # (not die) and latch once positions are readable — parity with barrier(). + # Without the guard the first raise kills the reader. + created: list = [] + table = make_table("unit.catchup.blip") + table._started = True + consumer = _fake_consumer(created, end_offset=1, position_errors=3)("unit.catchup.blip") + p0 = TopicPartition("unit.catchup.blip", 0) + task = asyncio.create_task(table._run(consumer, [p0], {p0: 1}, 0.0)) + task.add_done_callback(table._on_reader_done) + try: + assert await eventually(lambda: table.is_caught_up or table.status == "failed", timeout=2.0) + assert table.is_caught_up + assert table.status != "failed" + finally: + task.cancel() + # The reader may already be dead (boundary/pre-guard cases) or alive + # (post-guard latch); drain both the cancellation and any captured + # reader-death exception so cleanup never masks the assertions above. + with contextlib.suppress(asyncio.CancelledError, Exception): + await task + + async def test_catchup_gate_does_not_latch_during_position_blip(self) -> None: + # No false latch: while positions are unreadable the gate defers; it only + # latches on a complete successful read (the no-premature-latch property). + created: list = [] + table = make_table("unit.catchup.nolatch") + table._started = True + consumer = _fake_consumer(created, end_offset=1, position_errors=5)("unit.catchup.nolatch") + p0 = TopicPartition("unit.catchup.nolatch", 0) + task = asyncio.create_task(table._run(consumer, [p0], {p0: 1}, 0.0)) + task.add_done_callback(table._on_reader_done) + try: + assert await eventually(lambda: consumer.position_calls >= 3 or table.status == "failed", timeout=2.0) + assert not table.is_caught_up # deferred during the blip, not latched + assert table.status != "failed" # and not dead (pre-guard this fails fast) + assert await eventually(lambda: table.is_caught_up, timeout=2.0) # latches after recovery + finally: + task.cancel() + # The reader may already be dead (boundary/pre-guard cases) or alive + # (post-guard latch); drain both the cancellation and any captured + # reader-death exception so cleanup never masks the assertions above. + with contextlib.suppress(asyncio.CancelledError, Exception): + await task + + async def test_catchup_reraises_non_illegalstate_position_error(self) -> None: + # Boundary: the guard catches ONLY IllegalStateError. A different error + # from catch-up position() (a genuine broker fault) must still propagate + # and kill the reader — never be swallowed into an infinite catch-up spin. + # Mirrors barrier()'s narrow except. + created: list = [] + table = make_table("unit.catchup.fatal") + table._started = True + boom = KafkaError("genuine broker fault") + consumer = _fake_consumer(created, end_offset=1, position_errors=1, position_error=boom)("unit.catchup.fatal") + p0 = TopicPartition("unit.catchup.fatal", 0) + task = asyncio.create_task(table._run(consumer, [p0], {p0: 1}, 0.0)) + task.add_done_callback(table._on_reader_done) + try: + assert await eventually(lambda: table.status == "failed", timeout=2.0) + assert table.failure is boom + finally: + task.cancel() + with contextlib.suppress(asyncio.CancelledError, Exception): + await task + # --------------------------------------------------------------------------- # start() edge paths — real start() with induced broker conditions @@ -2034,17 +2158,23 @@ async def test_start_skips_ensure_topic_when_disabled(self, bootstrap, topic: st await writer.set("k", make_record("k", 1)) -async def test_raising_on_set_hook_kills_the_reader(topic: str, table_factory, writer_factory) -> None: - # End-to-end: a hook that raises in the reader loop is NOT swallowed. The - # unit test pins _apply's no-wrap decision; this confirms the consequence — - # the reader task dies and the failure surfaces as status 'failed'. +async def test_raising_on_set_hook_does_not_kill_the_reader(topic: str, table_factory, writer_factory) -> None: + # End-to-end: a hook that raises in the live reader loop is swallowed + + # counted, never fatal. The reader stays live, the base table still reflects + # the record, and stats.callback_errors records the skip. (Unit tests pin + # _apply's behavior; this confirms the consequence through the real loop.) def boom(key: str, value: object) -> None: raise RuntimeError("observer bug") async with writer_factory(topic) as writer, table_factory(topic, on_set=boom) as table: await writer.set("k", make_record("k", 1)) - assert await eventually(lambda: table.status == "failed") - assert isinstance(table.failure, RuntimeError) + assert await eventually(lambda: table.stats.callback_errors >= 1) + assert table.status != "failed" + assert table.failure is None + assert "k" in table # base record applied despite the raising hook + # the reader is still live: a later write is observed too + await writer.set("k2", make_record("k2", 2)) + assert await eventually(lambda: "k2" in table) # --------------------------------------------------------------------------- diff --git a/uv.lock b/uv.lock index 21cc7cc..276aa49 100644 --- a/uv.lock +++ b/uv.lock @@ -409,7 +409,7 @@ wheels = [ [[package]] name = "ktables" -version = "1.1.0" +version = "1.1.1" source = { editable = "." } dependencies = [ { name = "aiokafka" }, From e43f2453e7e03562a201c86e8b8a9d81068b18c8 Mon Sep 17 00:00:00 2001 From: Ryan Date: Fri, 3 Jul 2026 18:35:22 -0700 Subject: [PATCH 2/3] test: cover permanent-unassignment degrade and multi-partition catch-up blip Round-2 review follow-ups (no code-behavior change; tests + comment polish): - Add test_permanent_position_unavailability_degrades_not_fails: a permanent catch-up IllegalStateError (e.g. topic deleted mid-catch-up) defers to catchup_timeout -> degraded, never failed. The decision was documented but untested; guards against a future narrowing that would fail or hot-spin. - Add test_catchup_defers_whole_gate_on_single_partition_blip: locks the all-or-nothing catch-up gate semantic (one partition blipping defers the whole gate), deliberately distinct from barrier()'s per-tp skip. Adds a blip_partition param to the _fake_consumer helper. - Comment polish: retarget the ViewStats :meth: cross-ref to _notify_observer (where the logged/counted logic lives); drop gitignored spec-section refs from test comments; pin the timing invariant on the no-latch test. Full suite 254 passed, 100% branch coverage retained. --- ktables/kafka_table.py | 3 +- tests/test_kafka_table.py | 78 +++++++++++++++++++++++++++++++++++++-- 2 files changed, 76 insertions(+), 5 deletions(-) diff --git a/ktables/kafka_table.py b/ktables/kafka_table.py index 24f610e..50ee28b 100644 --- a/ktables/kafka_table.py +++ b/ktables/kafka_table.py @@ -455,7 +455,8 @@ class ViewStats: ``callback_errors`` counts times an ``on_set``/``on_delete`` apply-observer raised: the base table still applied the record, but the observer skipped it - (a raise is logged and counted, never fatal — see :meth:`KafkaTable._apply`). + (a raise is logged and counted, never fatal — see + :meth:`KafkaTable._notify_observer`). """ records_applied: int = 0 diff --git a/tests/test_kafka_table.py b/tests/test_kafka_table.py index df829dd..ce8d9ed 100644 --- a/tests/test_kafka_table.py +++ b/tests/test_kafka_table.py @@ -269,6 +269,7 @@ def _fake_consumer( stop_error: BaseException | None = None, position_errors: int = 0, position_error: BaseException | None = None, + blip_partition: int | None = None, ) -> type: """A configurable stand-in AIOKafkaConsumer for broker-free lifecycle tests. @@ -293,12 +294,17 @@ def _fake_consumer( raising ``IllegalStateError``; override the type with ``position_error`` (e.g. a ``KafkaError`` to prove the catch-up guard does NOT swallow it). The live ``position_calls`` counter lets a test observe the blip window. + - ``blip_partition``: if set, ONLY that partition's first ``position_errors`` + calls raise (others always succeed) — for the multi-partition case where one + tp blips while the rest are readable. Default (None) blips the first + ``position_errors`` calls regardless of partition. """ class _FakeConsumer: def __init__(self, *args: object, **kwargs: object) -> None: self.stopped = False self.position_calls = 0 + self._blip_calls = 0 topic = args[0] if args else kwargs.get("topic", "t") self._tps = {TopicPartition(str(topic), p) for p in partitions} self._fetch_max_wait_ms = kwargs.get("fetch_max_wait_ms") @@ -328,8 +334,16 @@ async def position(self, tp: object) -> int: # Caught up once position reaches the end offset; catch_up=False pins # it below so the reader loops indefinitely (degraded/cancellation). self.position_calls += 1 - if self.position_calls <= position_errors: - raise position_error if position_error is not None else IllegalStateError(f"Partition {tp} is not assigned") + if position_errors: + if blip_partition is None: + should_raise = self.position_calls <= position_errors + elif getattr(tp, "partition", None) == blip_partition: + self._blip_calls += 1 + should_raise = self._blip_calls <= position_errors + else: + should_raise = False + if should_raise: + raise position_error if position_error is not None else IllegalStateError(f"Partition {tp} is not assigned") return end_offset if catch_up else 0 async def stop(self) -> None: @@ -1813,7 +1827,7 @@ def bad_key(b: bytes) -> str: class TestApplyHooks: """on_set/on_delete fire on _apply's two branches, never on a rejected record; a raising hook is swallowed + counted (reader-liveness backstop for - external observers — see spec §4.1). All broker-free.""" + external observers). All broker-free.""" @staticmethod def _bytes_table(**kw: object) -> KafkaTable[bytes]: @@ -1883,7 +1897,7 @@ def bad_value(b: bytes) -> object: def test_apply_survives_raising_on_set_hook(self) -> None: # A raising on_set must NOT kill the reader: the base record is still # applied, the exception is swallowed and counted. Backstop for external - # observers; derived views stay compute-then-commit. See spec §4.1. + # observers; derived views stay compute-then-commit. def boom(k: str, v: bytes) -> None: raise RuntimeError("observer bug") @@ -2042,6 +2056,11 @@ async def test_catchup_gate_does_not_latch_during_position_blip(self) -> None: created: list = [] table = make_table("unit.catchup.nolatch") table._started = True + # Race-free by construction: the checkpoint (3) sits inside the blip window + # (position_errors=5), and each reader iteration yields once via getmany's + # 0.01s sleep while eventually() polls every 0.005s — so it cannot skip past + # the checkpoint between polls. Keep checkpoint < position_errors and the + # poll interval < the getmany yield if either is ever retuned. consumer = _fake_consumer(created, end_offset=1, position_errors=5)("unit.catchup.nolatch") p0 = TopicPartition("unit.catchup.nolatch", 0) task = asyncio.create_task(table._run(consumer, [p0], {p0: 1}, 0.0)) @@ -2080,6 +2099,33 @@ async def test_catchup_reraises_non_illegalstate_position_error(self) -> None: with contextlib.suppress(asyncio.CancelledError, Exception): await task + async def test_catchup_defers_whole_gate_on_single_partition_blip(self) -> None: + # All-or-nothing catch-up gate (deliberately distinct from barrier()'s + # per-tp skip): when ONE partition's position() blips, the WHOLE gate + # defers — it must not latch on the readable partition alone — then latches + # once every partition reads. Same race-free construction as the single- + # partition blip test (checkpoint 3 < blip window 5; poll < getmany yield). + created: list = [] + table = make_table("unit.catchup.multi") + table._started = True + p0 = TopicPartition("unit.catchup.multi", 0) + p1 = TopicPartition("unit.catchup.multi", 1) + # p1 blips for its first 5 position() reads; p0 is readable throughout. + consumer = _fake_consumer(created, partitions=(0, 1), end_offset=1, position_errors=5, blip_partition=1)("unit.catchup.multi") + task = asyncio.create_task(table._run(consumer, [p0, p1], {p0: 1, p1: 1}, 0.0)) + task.add_done_callback(table._on_reader_done) + try: + # while p1 blips, the gate defers — never latching on p0 alone + assert await eventually(lambda: consumer._blip_calls >= 3 or table.status == "failed", timeout=2.0) + assert not table.is_caught_up + assert table.status != "failed" + # once p1 recovers, both partitions read and the gate latches + assert await eventually(lambda: table.is_caught_up, timeout=2.0) + finally: + task.cancel() + with contextlib.suppress(asyncio.CancelledError, Exception): + await task + # --------------------------------------------------------------------------- # start() edge paths — real start() with induced broker conditions @@ -2126,6 +2172,30 @@ async def inflated(self, partitions, *args, **kwargs): async with table: assert table.status == "degraded" + async def test_permanent_position_unavailability_degrades_not_fails(self, monkeypatch: pytest.MonkeyPatch) -> None: + # §11 Decision A: a PERMANENT catch-up blip (position() always raises + # IllegalStateError — e.g. the topic is deleted mid-catch-up) must defer + # forever, reach catchup_timeout, and surface as 'degraded' (alive, + # recoverable) — NEVER 'failed'. Guards against a future narrowing of the + # catch-up guard that would send a permanent blip to failed (or spin). + # Broker-free: the fake consumer never lets position() recover. + created: list = [] + monkeypatch.setattr( + "ktables.kafka_table.AIOKafkaConsumer", + _fake_consumer(created, end_offset=1, position_errors=10**9), # never recovers + ) + table = KafkaTable.json( + bootstrap_servers=BOOTSTRAP, + topic="unit.catchup.permanent", + model=ServiceRecord, + ensure_topic=False, + catchup_timeout=0.2, + ) + async with table: + assert table.status == "degraded" # deferred forever -> degraded + assert table.failure is None # NOT failed: the guard kept the reader alive + assert created[0].stopped is False # still serving (torn down only on exit) + # --------------------------------------------------------------------------- # KafkaTableWriter — repr, lifecycle guards, ensure-topic toggle From d6e94e2b76212339494dca4c9fe1bb64a42f86aa Mon Sep 17 00:00:00 2001 From: Ryan Date: Fri, 3 Jul 2026 18:43:25 -0700 Subject: [PATCH 3/3] docs: add ADR 0003 for catch-up gate transient-unassignment tolerance --- ...p-gate-tolerates-transient-unassignment.md | 68 +++++++++++++++++++ 1 file changed, 68 insertions(+) create mode 100644 docs/adr/0003-catch-up-gate-tolerates-transient-unassignment.md diff --git a/docs/adr/0003-catch-up-gate-tolerates-transient-unassignment.md b/docs/adr/0003-catch-up-gate-tolerates-transient-unassignment.md new file mode 100644 index 0000000..bcaea0e --- /dev/null +++ b/docs/adr/0003-catch-up-gate-tolerates-transient-unassignment.md @@ -0,0 +1,68 @@ +# Catch-up gate tolerates a transient partition unassignment instead of failing the reader + +**Status:** accepted + +Under `group_id=None`, partition assignment is not fixed: `NoGroupCoordinator` +re-derives it on every metadata update. A metadata blip — most commonly a +`LeaderNotAvailableError` during a leader election or broker restart — momentarily +drops the subscribed topic from cluster metadata, so the assignment is briefly +replaced with an empty one and `consumer.position(tp)` raises `IllegalStateError` +until the next metadata refresh restores it. The reader loop's catch-up gate read +(`positions = [await consumer.position(tp) for tp in tps]`) was unguarded, so a +*self-healing* blip that landed while the table was still catching up killed the +reader permanently (`status == "failed"`, frozen for the object's lifetime). The +`barrier()` sweep already tolerated this exact exception; the catch-up gate did +not. We make the catch-up gate tolerate it too: catch `IllegalStateError`, defer +the gate for that iteration (`positions = None`), and latch only on a complete +successful read. aiokafka retries the blip away and the reassigned partitions +replay from earliest, so the fixed start-time end-offset gate re-evaluates and +latches on a later pass. + +## Considered options + +- **Leave the catch-up read unguarded (fail-fast)** — rejected: it turns a + transient, self-healing condition into a permanent, unrecoverable death, and it + is inconsistent with the `barrier()` path, which already defers on the very same + `IllegalStateError` (with a comment calling it "a blip aiokafka self-heals + from"). A sibling read guarding an exception that its twin lets kill the reader + is an oversight, not a decision. +- **Distinguish transient from permanent unassignment (bounded defer, then fail)** + — rejected: aiokafka gives no signal at the point of the raise to tell a + momentary blip from a permanent unassignment (e.g. a deleted topic). The + existing `catchup_timeout` already bounds an indefinite defer and converts it + into a loud `degraded` state, which is a better terminal than `failed` for a + condition that may still recover. +- **Broaden the catch to `KafkaError`/`Exception`** — rejected as dangerous: + `IllegalStateError` is the *only* self-healing signal here. Anything else + (including a genuine broker fault surfaced through `position()`'s + `check_errors()`) must still kill the reader loudly. Broadening would convert a + real fault into an infinite, silent `degraded` spin. + +## Consequences + +- **Transient blip mid-catch-up now recovers** instead of failing: the gate + defers, the assignment self-heals, and the reader latches on a later iteration. +- **Permanent unassignment now degrades, not fails.** A permanently-raising + `position()` (e.g. the topic is deleted mid-catch-up) defers forever, reaches + `catchup_timeout`, and surfaces as `status == "degraded"` (alive, recoverable, + loudly logged once at `start()`), never `failed`. This qualifies the library's + "non-retriable error latches `failed`" posture: a non-retriable *fetch* error + (e.g. `TopicAuthorizationFailedError`) still fails, but a permanent *catch-up + unassignment* degrades — the same defer-not-fail trade-off `barrier()` makes. +- **No false-positive catch-up.** Deferring sets `positions = None`, and the latch + is gated on `positions is not None and all(...)`, so the gate can only latch on + a complete successful read — never on partial or blipped data. The gate stays + all-or-nothing across partitions (one partition blipping defers the whole gate), + deliberately distinct from `barrier()`'s per-`tp` skip. +- **A genuine fault still dies loudly.** A non-`IllegalStateError` from the + catch-up `position()` propagates out of the reader task and latches `failed`. +- **Observability trade-off:** the `degraded` transition is logged once at ERROR + by `start()` and is queryable via `status`/`stats`; there is no periodic re-log + while permanently wedged. Accepted as-is — it matches the pre-existing + serve-degraded-don't-crash-loop posture and keeps the wedged state honestly + reflected in `status`. +- **Verified** by broker-free unit tests driving the reader loop with a fake + consumer whose `position()` raises `IllegalStateError` transiently (recovers to + `caught_up`), permanently (reaches `degraded`, never `failed`), and raises a + non-`IllegalStateError` (still `failed`); plus a multi-partition partial-blip + test pinning the all-or-nothing gate.