fix: reader survives raising apply-observer callbacks and transient catch-up unassignment#34
Merged
Merged
Conversation
…atch-up unassignment 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.
Contributor
Coverage report
Click to see where and how coverage changed
This report was generated by python-coverage-comment-action |
||||||||||||||||||||||||||||||
…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.
ryan-yuuu
pushed a commit
that referenced
this pull request
Jul 4, 2026
🤖 I have created a release *beep* *boop* --- ## [1.1.2](v1.1.1...v1.1.2) (2026-07-04) ### Bug Fixes * reader survives raising apply-observer callbacks and transient catch-up unassignment ([#34](#34)) ([cf3b040](cf3b040)) --- This PR was generated with [Release Please](https://github.com/googleapis/release-please). See [documentation](https://github.com/googleapis/release-please#release-please). Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Follow-up to the #33 investigation. Empirically, a broker disconnect does not kill the reader —
getmany()returns empty and aiokafka retries transport failures (verified against a real broker on a confirmed-silent port and against aiokafka 0.14.0 source). But the underlying "any exception out of the reader loop → permanentfailed, no recovery" fragility is real. This PR closes the two realistic in-loop exception sources.Fix #1 — a raising apply-observer callback no longer kills the reader
on_set/on_deletewere invoked unguarded in_apply, so any exception a callback threw latched the tablefailedfor its whole lifetime. Now a raise is logged (with the record key + traceback) and counted via the newViewStats.callback_errors; the base table still applies the record.KafkaTable's contract; a derived view's integrity is the view's own responsibility (compute-then-commit — whichGroupedKafkaTable's hooks already do, so this is purely an external-callback backstop).except Exceptiononly —BaseException/cancellation still propagates.Fix #3 — catch-up gate tolerates a transient
IllegalStateErrorUnder
group_id=None, a metadata blip (e.g.LeaderNotAvailableduring a leader election) can momentarily empty the assignment, makingposition()raiseIllegalStateError. Thebarrier()sweep already guards this; the catch-up gate did not, so a self-healing blip mid-catch-up was fatal. Now it defers the gate (mirrorsbarrier()) and latches only on a complete successful read.IllegalStateErrorstill propagates — a real fault must die, not spin (pinned by a boundary test).catchup_timeout→degraded(recoverable) rather thanfailed— same trade-offbarrier()already makes.Out of scope
restart()(would recover from any death, incl. non-retriable fetch errors like auth).Tests
TDD throughout. Full suite 252 passed (integration included), 100% branch coverage on
kafka_table.pyandgrouped_table.py. New coverage: on_set/on_delete survival + counter + key-in-log +BaseExceptionboundary; catch-up transient-blip survival, no-premature-latch, and the non-IllegalStateErrorboundary.Refs #33.