feat(core) astubbs#225: keep every completed record until the commit that carries it succeeds, and be able to put an aborted transaction's work back - #474
Conversation
…rries it succeeds, and be able to put an aborted transaction's work back The second rung under producer recovery, cut out of #410: the exactly-once argument, reviewable on its own. When a transactional producer is invalidated, the transaction that carried its output is aborted, so every record produced into it must be processed again - but by then the work manager has forgotten those records, because completion drops them. This rung keeps them. UncommittedCompletions is the ledger: what PartitionState.onSuccess would otherwise drop, kept with its offset until the commit carrying it succeeds, in transactional commit mode only (the no-op kind elsewhere, chosen once at construction). Its own monitor and its own thread-safety invariant, so the partition state's analysis does not inherit thirty accessors. The commit path trims only what the commit carried: the same collect-then-trim guard the incomplete set has, so a completion landing between collection and commit success survives for a later replay. The restore path puts an aborted transaction's work back: PartitionState.restoreCompletedButUncommittedWork re-registers each record as a fresh container at the partition's current epoch, register-then-publish in the order #370 established (#450), so a scanner completing the container the instant it is reachable finds its offset already in the incomplete set; the commit frontier drops below the restored offsets and the partition is marked dirty. PartitionStateManager and WorkManager carry it up, confined to the control thread and declared so. The processor's step is drain-then-replay: land every result already in the mailbox, so the ledger is complete, then replay. The drain itself is hardened for it - a successful-work listener throwing used to leave the results behind it in the drained batch unreachable for good; every result is now landed before the first failure is rethrown. The step asserts its confinement at the entry point, and nothing on this rung calls it: recovery, which does, is the PR above. User-visible: transactional mode now retains completed records until their commit, so memory held per partition grows with commitInterval; the README says so. Tests: the ledger through the state machine (replay restores what completed after the last commit and nothing before it, an empty ledger is a no-op, nothing is retained outside transactional mode, a key-ordered shard removed after the success is recreated, and the completion-on-publish pin), and the drain-then-replay order at the processor. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VrpH51xNDodaajE4P2nhFg
Dependency Review✅ No vulnerabilities or license issues or OpenSSF Scorecard issues found.Scanned FilesNone |
✅ Duplicate Code ReportTwo engines run in parallel for cross-validation. Each has its own thresholds tuned to its baseline - the real safety net is the per-engine "max increase vs base" check. ✅ PMD CPD
No new clones introduced by this PR. ✅ jscpd (language-agnostic)
|
…producer-fencing-brainstorm This PR now stacks on the ledger rung (#474), itself on the groundwork rung (#472), both cut out of this branch's tree today. Resolved by hand: the processor and the module resolve to this branch's versions, which are the rungs' text plus recovery; the README template takes the ledger rung's new retention section and points the recovery section's retention sentence at it; a restore entry point the merge placed twice in WorkManager is kept once. The ledger rung also carries two master commits this branch had not merged yet (the rebalance-callback rule and the Lincheck shard harness), which arrive with it. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VrpH51xNDodaajE4P2nhFg
|
[superseded - the quarantine lane is now empty] 🧪🔒 Quarantine Lane ReportThe superseded report, collapsed because it no longer applies
🔴 expected while the owner PR is open · 🟡🎲 flapper, pass proves nothing · 🚨 a deterministic quarantined test passing means its fix landed: delete its Updated for Superseded by a newer quarantine lane report. |
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## feat/225-recovery-groundwork #474 +/- ##
==================================================================
+ Coverage 83.21% 83.30% +0.08%
- Complexity 1622 1639 +17
==================================================================
Files 101 102 +1
Lines 5488 5559 +71
Branches 551 557 +6
==================================================================
+ Hits 4567 4631 +64
- Misses 728 733 +5
- Partials 193 195 +2
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
🟢 Throughput — OKThis branch measured about 16% faster than master, on the one test this measures. That is INSIDE this test's own run-to-run spread of about 17%, so read it as a reading and not as a result - re-running the same commit moves it by about as much.
Allowable range 🟢 ≥ 0.70 · 🟡 0.50–0.70 (about a 30% loss) · 🔴 < 0.50 (about a 50% loss) What the numbers mean, and what they cannot tell youThe one that gets misread. Why a shape and not a rate. A rate depends on which runner you drew. A shape does not: every test here processes a fixed number of records, so a runner twice as slow doubles the subject and the controls together and leaves their ratio alone. That is the whole trick, and it is why the reported rate is shown last and labelled as this machine only. Reading the comparison. By conservation, not by correction. Every test in this lane processes a fixed number of records, so within one run the ratio of one test's time to another's is invariant under machine speed — a runner twice as slow doubles both terms and leaves the ratio alone. There is no machine-index correction to be wrong, because nothing needed correcting. Per-method times, not class times. A class time is Reference is the median of 10 recent What this still cannot do. It removes machine-to-machine variance. It does not remove this test's own run-to-run variance, measured at about 30% on a single unchanged commit while its controls stayed within 5%. That is a property of the test, not of the comparison, and no arithmetic here can touch it — which is why the reference is a median and the bounds are deliberately coarse. 🟡 means look at this; only 🔴 is outside the measured spread. Runs used: e8bd2cb, 1743297, 4bc6e7a, b62c310, c381310, c79424a, 9c67c89, f1aa5eb, eb9fdb0, 51d9bb2 Since the previous push: ratio 0.988 -> 1.157, share 1.872 -> 1.599, rate 68676 -> 95526 (+39.1%). One push of difference sits inside this test's measured spread - read it as movement, not as a result. Updated for |
…nder it grew to five ab86e34 rewrote the list in this section for today's re-cut - #472 and #474 cut out from under this PR - and left the heading above it saying "three PRs (2026-09-03)". A reader who trusts the heading over the list it introduces gets the stack's shape wrong, and this note exists precisely to tell a later PR which rung it has to reconcile against. Verified against GitHub rather than the list: #426 MERGED, then #472 -> #474 -> #410 -> #420 all open, each based on the one before it. Five rungs, one landed, which is what the heading now says. The list's own contents were already correct and are untouched. Nothing enforces a heading against the block it introduces; grepped the tree and every ref first, as AGENTS.md requires before moving a labelled block - the string is cited nowhere, it only appears as this heading on the branches that carry their own copy of the note. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VrpH51xNDodaajE4P2nhFg
|
@claude review this |
|
Claude finished @astubbs's task in 3m 43s —— View job Review
This is a clean, well-isolated rung. I read the diff against
|
…thdrawn, rejoin is expressible Two of the burn-down's open items closed on 2026-09-08, each by its own agent, and the note records both as things that happened rather than things to do. The eager-mode PERIODIC_CONSUMER_SYNC stall (the confluentinc#857 family's fourth open item) is withdrawn by #478: four replays of the recorded seed on today's master drained completely with zero loss, and the "stall" was the Class 2 timing bound, whose crossing flips with the processor count at a fixed seed and tree. The grid it was opened on was also never a one-term A/B. Nothing to ship or to name; what survives is the per-shard liveness gap, already tracked and deferred. Whether "rejoin" after producer fencing is expressible in the lifecycle is known by a read of the #472/#474/#410 diffs against the engine's ownership rules: it is, and the question dissolves, because the instance never leaves the group - rejoin reduces to abort, replace and replay on the control thread under the write lock. The stack respects the @GuardedBy ledger, thread confinement and the lock pair. What it leaves is review-sized: #420's territory, one wire-level test nobody wrote, and the plan's one open question about declining the lock during a rebalance. None of it changes the tier 2 decision. Claude-Session: 460f7df9-dcc2-4b00-a9f9-62f3a2c6d5e4 Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
…-completions-ledger Brings the review fix on #472 (the start-up producer build no longer runs as user code) and the master it merged, which redesigned the two places the ledger hooks into: - #469 replaced the dirty flag pair with a completion count. The ledger snapshot now sits beside the count stash in getCommitDataIfDirty, and the restore path stamps recordCompletion() where it set the dirty flag, so the lowered commit frontier is re-offered. - #470 marks a partition clean only on the acknowledgement of the last offer it made. The ledger trims on the same condition and never on an older answer, since by then the snapshot is the newer offer's. - #466 gave processWorkCompleteMailBox its own control-thread assertion and a wake-up action; the ledger's drain hardening keeps its try/catch around both, and its duplicate assertOnControlThread helper is dropped for master's. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VrpH51xNDodaajE4P2nhFg
🧪🔒 Quarantine Lane ReportThe quarantine lane is empty - no Any earlier row on this PR asking for a Lane: non-gating; rules: see the Quarantine Audit check. No quarantined test changed outcome since the previous push. Updated for |
…-completions-ledger # Conflicts: # parallel-consumer-core/src/main/java/bz/stub/parallelconsumer/state/PartitionState.java
…ke, on the ledger rung The Integration Tests lane on this branch failed once at 779b280 on ManagedPCInstanceLifecycleTest's rapid-toggle case, the assertion the ledger already records. Codecov's per-commit history shows the same test passing at the branch's previous head and at the base branch's head in the same half hour on the same master, so this branch is not the variable. Recorded as a sighting; the fixed-sleep reading already on the ledger stands, and nothing is retried. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VrpH51xNDodaajE4P2nhFg
1 participant
Part of #225 - the second rung under recovery, cut out of #410 on 2026-09-07. The stack: #472 (vocabulary and plumbing), this PR (the ledger), then recovery itself (#410), then the rest of producer ownership (#420).
depends on #472
Description
This is the exactly-once argument of recovery, on its own. When a transactional producer is invalidated, the transaction that carried its output is aborted, so every record produced into it must be processed again - but completion had already dropped those records from the work manager. This rung keeps them, and can put them back.
The ledger.
UncommittedCompletionsholds whatPartitionState.onSuccesswould otherwise drop, with its offset, until the commit that carries it succeeds. Transactional commit mode only; the no-op kind everywhere else, chosen once at construction. It has its own monitor and thread-safety invariant, and the commit path trims only what the commit carried, so a completion landing between collection and commit success survives for a later replay.The restore path.
restoreCompletedButUncommittedWorkre-registers each record as a fresh container at the partition's current epoch, register-then-publish in the order #370 established and #450 asked this branch to match, so a scanner that completes a container the instant it is reachable finds its offset already registered. The commit frontier drops below the restored offsets and the partition is marked dirty.PartitionStateManagerandWorkManagercarry it up, confined to the control thread and declared so.The processor's step: drain, then replay - every mailboxed result landed first so the ledger is complete. The drain is hardened for it: a throwing successful-work listener used to strand the results behind it. The entry point asserts the confinement it declares. Nothing on this rung calls it; the recovery pass that does is #410.
User-visible: transactional mode retains completed records until their commit, so memory held per partition grows with
commitInterval; the README's new section says so.The one analysis finding attributed to this diff is inherited
bin/check-pr-analysis-surfaces.sh 474reports a single finding "on lines this PR wrote": SpotBugsNP_NONNULL_FIELD_NOT_INITIALIZED_IN_CONSTRUCTORforincompleteOffsetsatPartitionState.java:225. It is not this PR's. SpotBugs reports the bug at the constructor's declaration,incompleteOffsetshas always been assigned ininitStateFromOffsetDatarather than inline, and the identical annotation stands on the base branch's own run (#472,PartitionState.java:210) - this PR's field and assignment simply pushed that declaration down into its own hunk. Nothing to fix here; the pattern belongs to the SpotBugs registry.Proven able to fail
PartitionStateAbortedTransactionReplayTest: replay restores exactly what completed after the last successful commit, an empty ledger is a no-op, nothing is retained outside transactional mode, a key-ordered shard removed after the success is recreated, and the completion-on-publish pin (red on the publish-then-register order, observed before the swap).AbortedTransactionReplayStepTestpins the drain-then-replay order at the processor. Core suite green locally; every gate green.Checklist
docs/features/-N/A - a retention behaviour of transactional mode, documented in the README's transactional sectiondocs/inflight/working note (pr-/branch-) started at the PR's first commit -docs/inflight/core-recoverable-producer-fencing.mdcovers the whole of Producer fencing kills the instance; it should abort and rejoin, like Kafka Streams' TaskMigratedException #225 and lists the stackce-simplifyandce-code-reviewlocally -N/A - a strict subset of astubbs/parallel-consumer#410, which had both passes, an automated review and a Codex review before the cut; the ledger's own extraction was one of that review's findings🤖 Generated with Claude Code
https://claude.ai/code/session_01VrpH51xNDodaajE4P2nhFg