Repository navigation
fix(core) astubbs#248: an async commit counts as committed when the broker answers, not when it is sent - #470
Merged
Conversation
…ker answers Two failing tests for the bug recorded in docs/inflight/bug-async-commit-marked-successful-before-broker-ack.md: under the shipped-default PERIODIC_CONSUMER_ASYNCHRONOUS commit mode, an offset is treated as committed at the point commitAsync is SENT, not when the broker acknowledges it through the OffsetCommitCallback. The two ways a sent request never becomes a durable commit get one test each, because they are one scenario with one variable: - MockConsumerAsyncCommitCallbackFailsTest - the callback arrives carrying a RetriableCommitFailedException, the failure commitAsync is specified to report. - MockConsumerAsyncCommitCallbackDroppedTest - the callback never arrives at all, which is the harder case because there is no event to hang a decision on. The discriminating assertion is the BROKER's committed offset, not "no exception escaped". All ten records are published before processing starts, so they are complete before the first commit falls due and that commit carries the whole batch. Marking clean on send therefore ends the story: nothing is dirty afterwards, collectCommitDataForDirtyPartitions() returns empty, no further commitAsync is issued, and the mock consumer's committed() stays empty forever while PC believes offset 10 is durable. Silent loss - the next owner of the partition resumes from a position the broker never recorded. Red proof, both tests, on this branch before any main-code change: expected to be greater than: 3 but was : 2 within 30 seconds Commit attempts stall at two of the three withheld acknowledgements and stop, and the broker-side assertion is never reached. The prediction stated before the run was "attempts stop before the withheld window ends"; it held. Reuses MockConsumerTestBase rather than adding a harness, and mirrors CommitRejectionTestBase - its sibling family for the synchronous mode, where the rejection arrives as a thrown exception instead of a callback. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019SVm2cT6ZgUyPMim9CukYK
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)
|
![github-actions[bot]](https://avatars.githubusercontent.com/in/15368?s=60&v=4){kind=small}
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #470 +/- ##
============================================
+ Coverage 82.21% 82.57% +0.36%
- Complexity 1537 1551 +14
============================================
Files 96 96
Lines 5353 5373 +20
Branches 533 536 +3
============================================
+ Hits 4401 4437 +36
+ Misses 757 739 -18
- Partials 195 197 +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:
|
[superseded - a quarantined test changed outcome] 🧪🔒 Quarantine Lane Report
🔴 expected while the owner PR is open · 🟡🎲 flapper, pass proves nothing · 🚨 a deterministic quarantined test passing means its fix landed: delete its No quarantined test changed outcome since the previous push. Updated for Superseded by a newer quarantine lane report. |
🟢 Throughput — OKThis branch measured about 11% 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: 6aab3ff, 745b1f6, 49f81f1, bb73082, 16b2e89, ed82e7f, c562df4, f502202, abd1d39, 60138dc Since the previous push: ratio 0.975 -> 1.114, share 1.744 -> 1.527, rate 68246 -> 72472 (+6.2%). One push of difference sits inside this test's measured spread - read it as movement, not as a result. Updated for |
|
…hen it is sent
Under PERIODIC_CONSUMER_ASYNCHRONOUS - the shipped default commit mode -
AbstractOffsetCommitter marked the partition clean on the line after
commitOffsets() returned. commitSync and the transactional producer return only
once the broker has answered, so that marking is a statement about something that
already happened. commitAsync does not: it returns as soon as the request is
handed to the client and answers later through an OffsetCommitCallback. So the
offsets were recorded as durable at the moment of SENDING.
What that cost, precisely. Marking clean ends the story: nothing is dirty
afterwards, collectCommitDataForDirtyPartitions() returns empty, and no further
commitAsync is ever issued for those offsets. A callback then arriving with a
RetriableCommitFailedException had nothing left to retry - "retriable" described a
retry nothing would perform - and a callback that never arrived was
indistinguishable from success. The broker's committed offset stayed behind PC's
belief, silently and permanently, so whoever owned the partition next resumed from
a position the broker never recorded.
The fix, two pieces and deliberately minimal:
- AbstractOffsetCommitter gained commitOffsetsReturnsOnlyOnceAcknowledged(), true
by default, and marks success inline only when it holds. onOffsetCommitSuccess
became protected so a committer answering false can call it on its own
acknowledgement.
- ConsumerOffsetCommitter returns isSync() from it, and its async branch registers
onAsyncCommitAnswered: success marks clean, failure logs and leaves the offsets
dirty for the next cycle to re-send, a dropped callback does the same by doing
nothing at all.
A failure is now a DEFERRAL and lands where the synchronous path's rejections
already land - logged, not fatal, offsets still marked as needing a commit. That is
the option commitDeferringOnRebalance()'s javadoc argues for at length, reached
from the other direction: there, deferral is what a thrown exception causes by
aborting before the success marking; here it is what the absence of a success
signal causes. Same rule - only an acknowledgement advances the state.
One hazard the fix itself creates, and its guard. Deferring the clean-marking is
what makes two async commits able to be in flight at once; before, the first send
marked clean so there was never a second. An acknowledgement of a SUPERSEDED
request must not mark clean, because the newer request carries higher offsets and
is still unanswered. ConsumerOffsetCommitter numbers its sends and the callback
ignores any answer that is not the latest - the monotonic-sequence discipline
commitAsync's own javadoc recommends. It costs at most one extra commit and cannot
under-report.
Evidence. The two red tests from the previous commit go green, and both control
arms flip exactly one term - the new predicate - and nothing else:
hook forced true (pre-fix) both async tests fail: expected to be greater
than: 3 but was: 2; sync control passes
hook = isSync() (shipped) all three pass
hook forced false sync control fails: expected: 2 but was: 54;
async tests pass
The third arm is why MockConsumerSyncCommitMarksCleanOnReturnTest exists. It
asserts that commit attempts STAND STILL once the batch is committed, not merely
that the offset was reached - which is true either way on the first commit. A sync
mode that stopped marking clean would re-commit every commitInterval forever, and
only the attempt count says so.
ConsumerOffsetCommitterAsyncFailureLoggingTest's line-length bound moved, and this
is the one judgement call worth reviewing. Its constant covers the log statement's
own fixed text, which roughly doubled because the ERROR line now states what the
failure MEANS for the offsets. The per-entry term - the half that guards
#168 / confluentinc#629, the offset map being interpolated back in at
DefaultMaxMetadataSize per partition - is untouched, and the measured line sits at
296 against a 380 bound while the defect it catches would add 4096.
Other instances of the class - state advanced on request rather than on
acknowledgement - checked and dismissed. The produce path returns the send futures
and blocks on futureSend.get(sendTimeout) before the batch is marked succeeded, and
a failure throws. ProducerManager's transactional commit is blocking, so the base
class's default is right for it. Vert.x, Reactor and Mutiny all mark
onUserFunctionSuccess() from the completion signal, and ExternalEngine suppresses
the dispatch-time marking for exactly this reason. No further instances found.
The inflight note is retired and its mechanism migrated to
docs/solutions/logic-errors/an-async-commit-was-recorded-on-send-not-on-acknowledgement-2026-09-07.md;
bug-torn-read-family.md's citation of it now points there, in the shape that note
already uses for its other closed out-of-family finding. The upstream manifest
entry sweep-2023-broker-disconnect-commit is updated at the same time: it quoted
the very TODO this commit deletes, and it stays OPEN because the reporter's actual
symptom was a throughput collapse nothing here reproduced.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_019SVm2cT6ZgUyPMim9CukYK
astubbs
marked this pull request as ready for review
September 7, 2026 10:45
astubbs
marked this pull request as draft
September 7, 2026 10:56
Owner
Author
|
@claude review this |
|
Claude finished @astubbs's task in 4m 1s —— View job Review
SummaryThe fix is sound and the diagnosis in the PR body matches what's in the code: Two things worth a look before merge, plus one non-blocking observation: 1. No test exercises the sequence-number guard itself ( public void onOffsetCommitSuccess(OffsetAndMetadata committed) { //NOSONAR
lastCommittedOffset = committed.offset();
setClean();
}This assignment is unconditional — nothing downstream stops 2. if (exception != null) {
log.error("Async offset commit deferred, not dropped - these offsets stay dirty and are re-committed " + ...);
return;
}
long latest = asyncCommitSequence.get();
if (sequence != latest) { ... }If an older, already-superseded request's callback fails after a newer request has already succeeded, this still logs at ERROR claiming "these offsets stay dirty and are re-committed next cycle" — which may already be false (a higher offset was already committed and the partition may already be clean). No state is corrupted either way since the branch just returns, so this is a log-quality nit rather than a correctness bug, but it could generate a misleading ERROR for on-call/monitoring in that narrow race. Checking 3. Non-blocking: the bot's duplicate-code check flagged ~13-15 line overlaps between Everything else checks out: the PR's "other instances of this defect class" sweep (produce path, transactional commit, |
… the guard has a test Both findings from the automated review on #470, which read the sequence guard this branch added and found it undefended and mis-ordered. FINDING 1 - nothing exercised the guard, and nothing downstream backs it up. The review checked PartitionState.onOffsetCommitSuccess and found the assignment unconditional: lastCommittedOffset = committed.offset() and then setClean(). It neither compares against what it already holds nor knows another request is outstanding. So the sequence != latest check in ConsumerOffsetCommitter.onAsyncCommitAnswered is not belt-and-braces - it is the whole of the protection against a late acknowledgement of a superseded request walking the recorded commit offset BACKWARDS, and against marking a partition clean at an offset a newer, still-unanswered request had already passed. That is the original defect of this PR re-entered through the door the fix opened, since deferring the clean-marking is what makes two async commits able to be in flight at once at all. The MockConsumer scenario tests cannot reach it: MockConsumer.commitAsync answers inline on the calling thread before it returns, so a request is always already answered by the time the next one is sent and the overlap never exists. Reaching it needs the callback held, which needs it captured. New ConsumerOffsetCommitterSupersededAsyncCommitTest does that, with three cases - the older answer arriving while a newer request is outstanding, the out-of-order answer that would regress the offset, and the superseded failure's log line - asserting at the WorkManager boundary, which is the seam the committer's decision is visible at without standing up a partition-assigned engine. FINDING 2 - the failure branch logged before checking supersession. The ERROR line is a contract with whoever is on call: it says these offsets stay dirty and this cycle re-commits them. For a superseded request that claim can be false - a later request has already been sent, and its answer is what decides those offsets. Supersession is now checked first, so a superseded answer is reported as supersession whether it succeeded or failed. The failure half is a WARN naming the superseding sequence rather than a DEBUG, because a request really did fail; it is simply not the ERROR that promises a re-commit is in hand. The #168 (confluentinc#629) bound applies to the new line as much as to the old one, so it summarises the offsets rather than interpolating the map, and the test asserts that. RED PROOF - the guard removed, everything else identical, all three tests red, and each failure named its own mechanism rather than a generic assertion: aSupersededAcknowledgementIsIgnoredAndTheStillOutstandingOneIsNot NeverWantedButInvoked: workManager.onOffsetCommitSuccess(<any map>) ... but invoked here with offset=100 - the superseded answer marked the partition clean. anAnswerThatArrivesAfterAHigherOneCannotWalkTheCommittedOffsetBackwards expected: [200] but was: [200, 100] - the recorded offset went backwards, which is the direction of the regression and why the offsets are asserted rather than the call count. aSupersededFailureSaysItWasSupersededInsteadOfClaimingAReCommit expected to be empty but was: [Async offset commit deferred, not dropped - these offsets stay dirty and are re-committed next cycle ...] - the false promise, at ERROR. Restoring the guard turns all three green. Tests run: 3, Failures: 0. Also here, and not part of either finding: LogCapture gained messagesAt(level, mentioning...) and onlyMessageAt(level, mentioning...), which is the filter-on-something-unique obligation its own javadoc states made into API rather than re-implemented per test class, and AsyncCommitterFixture holds the setup both hand-driven committer tests need - the two mocks, an async-mode committer, the captured callbacks and a one-partition commit - so the second test class did not arrive as a copy of the first. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019SVm2cT6ZgUyPMim9CukYK
…et; only the clean mark waits for the highest one in flight The sequence guard the previous commit added ignored the answer to a superseded commitAsync request outright. That is the wrong shape, and the maintainer's call replaces it: a late acknowledgement is TRUE - the broker really did commit up to the offsets it names, for every partition it names - so dropping it throws away a fact the broker established. The only thing that must be withheld is the CLEAN mark, and only for a partition that still has a higher offset in flight, because that is the state that would leave nothing dirty to re-send if the newer request then failed or was dropped. THE RULE, PER PARTITION. ConsumerOffsetCommitter keeps highestOffsetInFlight, a map written on send (merged with Math::max, so it only rises while a request is outstanding) and cleared by the acknowledgement that matches it - an atomic compare-and-remove, because sends and answers are not always on the same thread. asyncCommitSequence and the sequence != latest check are gone. On an acknowledged success, each partition in the answer goes one of two ways: its offset IS the highest in flight record it, and mark the partition clean a newer request has passed it record it, and leave the partition dirty A request is routinely the newest word on one partition and superseded on another - the commit carries every dirty partition, and only the ones that completed more work move - which is why the decision cannot be per request. The recording half is PartitionState.onSupersededOffsetCommitSuccess, beside the existing onOffsetCommitSuccess; both now go through recordCommittedOffset, which advances lastCommittedOffset MONOTONICALLY. That guard lives where the field lives: the field is what pc.partition.latest.committed.offset reads, and an out-of-order answer would otherwise walk the gauge backwards. PartitionState's clean protocol is otherwise untouched - setClean still declines when state changed during the commit window, and the counter protocol on #469 is deliberately not pulled in. On a failure nothing moves, as before, and the line now says which promise it is making. ERROR - "these offsets stay dirty and are re-committed next cycle" - when the request still carries the highest in-flight offset for at least one partition, because for those partitions the promise is exactly true. WARN naming the supersession when it carries it for none, because then a newer request decides every offset on the line and the ERROR would be a claim this answer is not entitled to make. The #168 (confluentinc#629) bound applies to both: the offsets are summarised, never the map. Sync mode is untouched - commitOffsetsReturnsOnlyOnceAcknowledged() still returns isSync(), so the blocking path still marks success inline, and MockConsumerSyncCommitMarksCleanOnReturnTest stays green. RED PROOFS. The committer's decision rule, proved against the sequence guard still in place - the new WorkManager/PartitionState methods existed so the tests could compile, so the one term flipped is onAsyncCommitAnswered itself. Four of the five went red, each naming its own mechanism: anOlderAnswerIsRecordedAndLeavesThePartitionDirtyForTheHigherOffsetStillInFlight the broker committed up to 100, so that offset is recorded whether or not it is the newest word on the partition / expected: [100] but was: [] anAnswerThatArrivesAfterAHigherOneAppliesNothing expected: [100] but was: [] - the late answer was dropped instead of recorded theOlderAnswerMarksCleanTheOnePartitionTheNewerRequestDidNotSupersede Wanted but not invoked: workManager.onOffsetCommitSuccess({...-1 = offset=50 ...}); Actually, there were zero interactions with this mock - a whole-request guard cannot mark one partition clean and leave the other dirty aFailureStillCarryingTheHighestOffsetForOnePartitionKeepsTheReCommitPromise captured ERROR lines ... expected: 1 but was: 0 - reported as supersession although the promise was true for partition 1 The fifth, aFailureSupersededOnEveryPartitionSaysTheNewerRequestDecidesTheOutcome, passes under both rules; it is the previous commit's case, kept and reworded. PartitionStateAcknowledgedCommitOffsetTest asserts the other side of the seam on a real PartitionState - what each entry point does to the offset and the dirty flag - and was proved by sabotage, one term at a time: monotonic check removed (assign unconditionally) aLateAcknowledgementCannotWalkTheCommittedOffsetBackwards expected: 200 but was: 100 onSupersededOffsetCommitSuccess made to setClean() too aSupersededAcknowledgementRecordsItsOffsetAndLeavesThePartitionDirty expected to be true Both restored; all three green afterwards. ConsumerOffsetCommitterSupersededAsyncCommitTest is renamed to ConsumerOffsetCommitterOverlappingAsyncCommitTest - the subject is no longer a guard against superseded answers but what an overlapping pair of requests each moves - and AsyncCommitterFixture gained a two-partition commit, which is the only shape that distinguishes a per-partition rule from a whole-request one. WorkManager gained a three-line delegation beside its sibling rather than a new responsibility, so no decomposition was attempted here; the file's refactor candidacy is unchanged. Gates: bin/ci-unit-test.sh green across every module, bin/check-all.sh 16 passed 0 failed, bin/check-issue-refs.sh clean. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019SVm2cT6ZgUyPMim9CukYK
…ans it, so the committer keeps no in-flight map The maintainer's call on the previous commit, which duplicated state the partition already tracks. ConsumerOffsetCommitter kept its own per-partition highestOffsetInFlight map, split every acknowledgement into two buckets, and plumbed a second success method - onSupersededOffsetCommitSuccess - through AbstractOffsetCommitter, WorkManager, PartitionStateManager and PartitionState, to reach an outcome the partition produces on its own. Only the CLEAN mark ever needed guarding, and the partition knows the offset it offered. All of that is replaced by one rule. THE RULE. PartitionState.getCommitDataIfDirty remembers the offset it hands to the committer, in one new field, offsetLastOfferedForCommit. onOffsetCommitSuccess then records every acknowledgement - the answer is TRUE, the broker committed up to the offset it names, and recordCommittedOffset keeps the higher of the two when answers arrive out of order - and marks the partition clean ONLY when the offset acknowledged is that offer. setClean's existing protocol is otherwise untouched: it still declines when state changed during the commit window, and the counter protocol on #469 is deliberately not pulled in. commitOffsetsReturnsOnlyOnceAcknowledged() and the sync path are unchanged. The committer's async callback is now two branches. On success it passes the acknowledgement through whole. On failure it logs ONE line at WARN saying the partitions stay dirty and are committed when a later request is acknowledged, with the #168 (confluentinc#629) summarisation and the DEBUG line carrying the full map, and returns. The previous ERROR-or-WARN split existed only because the in-flight map could tell the two cases apart; nobody on call needed the distinction, and a routine coordinator hiccup is not an ERROR when nothing was lost and nothing was marked clean. Deleted: highestOffsetInFlight, recordAsInFlight, acknowledgeTheHighestOffsetInFlight, stillCarriesTheHighestOffsetInFlightForAnyPartition, and onSupersededOffsetCommitSuccess in all four classes. Added: one long. THE PER-PARTITION OUTCOME SURVIVES, AND NOW FALLS OUT. A commit carries every dirty partition and only the ones that completed more work move, so a request is routinely the newest word on one partition and superseded on another. In the second round the partition that stood still is re-offered at exactly the same offset, so the first round's answer IS its latest offer and cleans it, while the partition that moved on stays dirty. Nothing implements this; each partition made its own offer, so each partition decides. Every edge lands on the dirty side, which costs at most one extra commit and cannot under-report: a partition state rebuilt by a rebalance starts at KAFKA_OFFSET_ABSENCE, so an acknowledgement addressed to the assignment before it cannot clean it either - better than the previous unconditional setClean(). THE FIELD'S THREAD MODEL, per the engine's AGENTS.md. Written in the commit path (getCommitDataIfDirty, reached from collectCommitDataForDirtyPartitions) and read back in onOffsetCommitSuccess, by the SAME thread in every commit mode: the broker-poll thread under both consumer commit modes - it sends the request and Kafka delivers the callback from the poll it drives - and the control thread under PERIODIC_TRANSACTIONAL_PRODUCER, where the commit blocks and the acknowledgement is recorded inline. The one hand-over is Consumer#close() flushing a pending callback after the poll loop has finished, which carries a happens-before edge rather than an overlap. So it is plain, for the reason the dirty field's javadoc gives from jcstress measurements: that flag is this class's fence and the longs stay plain deliberately. Same lifecycle as stateChangedSinceCommitStart, and the non-volatile ledger in docs/refactoring.md now records the pair so neither is "fixed" with volatile and so they are declared together if anyone does the @ThreadConfined work. Assigned rather than raised to a maximum: the offer rises with completed work, so the last offer is the highest in flight, while a monotonic maximum would also survive maybeTruncateBelowOrAbove's downward reset on a bootstrap poll - after which the partition could never match its own offer again and would be dirty forever, committing nothing. RED PROOF - the PartitionState rule reverted (unconditional setClean() on success restored), nothing else touched. Tests run: 7, Failures: 4: ConsumerOffsetCommitterOverlappingAsyncCommitTest .anOlderAnswerLeavesThePartitionDirtyForTheHigherOfferStillOutstanding offer 2 is still unanswered, so the next cycle must still be willing to send it / expected: {...-0=2} but was: {} ConsumerOffsetCommitterOverlappingAsyncCommitTest .oneAnswerCleansThePartitionItIsTheLatestOfferForAndLeavesTheOtherDirty only the partition a newer offer is outstanding for may stay dirty / expected: {...-0=2} but was: {} PartitionStateAcknowledgedCommitOffsetTest .anOlderAcknowledgementAfterAHigherOfferRecordsTheOffsetAndLeavesThePartitionDirty offer 201 is still unanswered ... expected to be true PartitionStateAcknowledgedCommitOffsetTest .oneAnswerCleansThePartitionItIsTheLatestOfferForAndLeavesTheOtherDirty a newer offer for this partition is still unanswered ... expected to be true The empty map is the partition marked clean at the older offset while the higher offer is still outstanding - exactly the state that leaves nothing to re-send. Restored, all seven green. The out-of-order case passes under both rules, since the clean mark is already taken by the newer answer; it is kept because it pins the other half - the late answer must not put the partition BACK into a re-commit of offsets the broker already has. TESTS. PartitionStateAcknowledgedCommitOffsetTest pins the rule where it lives, on a real PartitionState: older ack after a higher offer records and does not clean; ack of the offered offset cleans; a lower ack after a higher recorded offset moves nothing backwards; and the mixed case on two partitions. Every test reads the offer back rather than naming an offset - the offered offset is derived from the incomplete set, so a hard-coded one would pass against a partition that offered something else entirely. ConsumerOffsetCommitterOverlappingAsyncCommitTest drives the same three orderings through the committer with two captured callbacks, now over a real WorkManager and real partition states rather than a WorkManager mock: with the decision inside PartitionState, asserting at the mock boundary would assert almost nothing. It asserts what reached the broker (the captured commitAsync maps) and what the next cycle would still send (collectCommitDataForDirtyPartitions - the consequence rather than the flag, and the exact call whose emptiness was the original defect). AsyncCommitterFixture now hands back each send's offsets and callback together, so a test says what reached the broker rather than assuming it. ConsumerOffsetCommitterAsyncFailureLoggingTest moves from ERROR to WARN and additionally asserts NO ERROR, so a false alarm reappearing on a routine coordinator hiccup fails the build. Its per-entry length bound - the half guarding #168 - is untouched. The four MockConsumer async and sync tests are unchanged and stay green. The static: analysis lane was RED on the previous head over a ratchet identity, NULLPTR_DEREFERENCE PartitionStateManager.lambda$onSupersededOffsetCommitSuccess$4. This deletes the method the finding was in rather than adding a ratchet line for it; bin/infer-test.sh now reports 10 findings, all known, none new. Gates: bin/ci-unit-test.sh green across every module, bin/check-all.sh 16 passed 0 failed, bin/check-issue-refs.sh and bin/check-file-refs.sh clean. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019SVm2cT6ZgUyPMim9CukYK
[superseded - a quarantined test changed outcome] 🧪🔒 Quarantine Lane Report
🔴 expected while the owner PR is open · 🟡🎲 flapper, pass proves nothing · 🚨 a deterministic quarantined test passing means its fix landed: delete its Since the previous push: Updated for Superseded by a newer quarantine lane report. |
astubbs
marked this pull request as ready for review
September 7, 2026 23:30
Owner
Author
|
@codex review |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: af758c662b
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
…perseded metadata cannot clean it The Codex review round on #470. Six of its seven P1 findings are addressed here; the seventh is declined in-thread and left for the maintainer. THE DEFECT THE REVIEW FOUND, AND WHY IT IS REAL. PartitionState compared an acknowledgement against offsetLastOfferedForCommit, a long. The offered offset is one above the highest SEQUENTIALLY succeeded offset, while the metadata riding with it is the encoded set of incomplete offsets above that - so completing a record above the lowest incomplete one changes the metadata and moves the offset NOT AT ALL. That is not an exotic case. It is the ordinary consequence of out-of-order completion, which is the thing this library exists to do. Two requests are then in flight carrying the same offset and different metadata; compared on offsets alone the answer to the OLDER one matches, marks the partition clean, and the newer request's metadata - the only record that the higher record is done - is never re-sent if that request fails or is dropped. Those records are re-delivered after a reassignment: not lost offsets, but precisely the replay the encoded offset map exists to prevent, and quieter than the defect this PR is about because the committed offset is correct throughout. THE FIX. One field changes type. offsetLastOfferedForCommit (long) becomes offerLastMadeForCommit (OffsetAndMetadata, null until the first offer), and onOffsetCommitSuccess compares with equals() instead of ==. Comparing whole is EXACT, not merely tighter: Consumer#commitAsync hands the OffsetCommitCallback the very map it was given, and the transactional path calls onOffsetCommitSuccess inline with the map it just collected, so an offer's acknowledgement carries that offer's own object. Anything not coming back identical leaves the partition dirty - one extra commit, cannot under-report, the same safe edge the rest of the rule already lands on. The "assigned, not raised to a maximum" paragraph is rewritten rather than dropped: the reason only the latest offer is kept is unchanged, but a maximum is no longer the alternative being rejected - a SET of outstanding offers is, and maybeTruncateBelowOrAbove's downward reset on a bootstrap poll is still the event no pruning rule would see. RED PROOF - new test, main code untouched, everything else green: PartitionStateAcknowledgedCommitOffsetTest .anAcknowledgementOfAnOfferWithSupersededMetadataAtTheSameOffsetDoesNotCleanThePartition the newer offer is still unanswered and is the only thing carrying the completion of record 300 - cleaning here would leave nothing dirty to re-send it / expected to be true Its two guard assertions both held first (the offers carry the SAME offset and DIFFERENT metadata), so the scenario is genuine rather than the ordinary superseded case wearing a new name. Tests run: 5, Failures: 1 before, 0 after. The four existing tests in that class now acknowledge by handing the offer object back, which is what the real callback does; reconstructing an OffsetAndMetadata from the offset alone tested a request shape that never occurs and was what let the metadata half go unasserted. offer() returns the offer whole for that reason. DUPLICATE-CODE REPORTS, both on this PR's own files. The bot flagged AsyncCommitAcknowledgementTestBase against CommitRejectionTestBase (15 lines) and MockConsumerSyncCommitMarksCleanOnReturnTest against the same file (13 lines). Three await-and-read blocks were verbatim copies down to the Collections.singleton(topicPartition) lookup, so they are hoisted into MockConsumerTestBase: awaitAllRecordsProcessed, awaitBrokerCommittedOffset and assertParallelConsumerStillRunningWithNoFailureCause. MockConsumerTestBase's javadoc said assertions are deliberately NOT hoisted, because their timeouts are scenario-specific. That decision is NARROWED, not reversed, and the javadoc now records the override where it is made: every helper takes its deadline as an argument, so each scenario still states its own timeout in its own file beside the assertion it bounds. What moved is only plumbing that carries no scenario-specific judgement. The commit-attempt assertion is deliberately NOT hoisted, although two of the three files spell it identically: the two rejection/withholding families assert it CLIMBING past a floor and the synchronous control asserts it STANDING STILL over a quiet period, which is the distinction those tests exist to draw. A shared helper with a predicate argument would have obscured it. That refactor was proved to have kept its teeth, by sabotage rather than by reading: commitOffsetsReturnsOnlyOnceAcknowledged() forced true (pre-fix behaviour) both async tests red through the hoisted path: expected to be greater than: 3 but was: 2 within 30 seconds; sync control green forced false (over-applied) sync control red: expected: 2 but was: 54; async tests green Both reproduce the numbers this PR's body already records, so the helpers assert what the inline blocks did. All eleven MockConsumerTestBase subclasses green, including the six this change does not touch. THE OTHER FINDINGS. Applied: the writer/reader/fence analysis duplicated into docs/refactoring.md is collapsed to the work item plus a pointer naming offerLastMadeForCommit's javadoc as the owner - two copies of a concurrency argument drift, and the field is where a reader forms the question. docs/inflight/issue-response-248.md is added, which docs/inflight/AGENTS.md requires of any note mapping to an issue before its PR merges and which this branch was missing; it says what is fixed, and that the reported throughput collapse is NOT explained, which is why #248 stays open. The subject of this commit carries the issue number, and these trailers carry the upstream provenance the manifest records. Declined, in-thread and left open for the maintainer: @ThreadConfined on the new field. The engine's AGENTS.md asks for the annotation plus a runtime assertion, but ThreadConfined.ANY would latch the broker-poll thread and then fire on the Consumer#close() callback flush - a legitimate hand-over the field's javadoc already documents - and docs/refactoring.md records that this field and stateChangedSinceCommitStart are one lifecycle to be declared together or not at all. Annotating one of the pair reads as a claim about the other. Declined as unactionable: renaming the branch to encode confluentinc#203. Renaming a branch with an open PR closes that PR, and the manifest's fork.branches records the current name. Gates: bin/ci-unit-test.sh green across every module; bin/check-all.sh 16 passed, 0 failed; bin/check-issue-refs.sh and bin/check-file-refs.sh clean. Upstream-Issue: confluentinc#203 Upstream-PR: confluentinc#345 Forwarded: no Applied-Upstream: no Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019SVm2cT6ZgUyPMim9CukYK
…ccess-waits-for-ack # Conflicts: # parallel-consumer-core/src/main/java/bz/stub/parallelconsumer/state/PartitionState.java
astubbs
changed the title
Owner
Author
|
@codex review |
🧪🔒 Quarantine Lane Report
🔴 expected while the owner PR is open · 🟡🎲 flapper, pass proves nothing · 🚨 a deterministic quarantined test passing means its fix landed: delete its No quarantined test changed outcome since the previous push. Updated for |
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 7389db5dfb
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
… no later request can fix it The second Codex round on #470, which raised one P1: the failure branch treats every exception as a transient deferral. Half of that is this PR's to fix and is fixed here; the other half is deliberately not, and the reasoning is recorded rather than left implied. WHAT WAS WRONG, AND THAT IT IS THIS PR'S DOING. The WARN this branch introduced promises that the partitions "stay dirty and are committed when a later request is acknowledged". For a coordinator hiccup that is true and an operator can go back to sleep. For a permanent failure - group authorization revoked, the instance fenced - it is FALSE: every later request fails identically, the partitions stay dirty for the life of the process, the committed offset never advances, and nothing improves without a person. That consequence is new. Before this branch the offsets were marked clean at send, so a permanent failure lost them silently and there was no retry to spin; now they are correctly retained, which is right, and the retry spins forever against a doomed commit while the only signal for it had been downgraded from ERROR to WARN by the previous commit. Reporting that at WARN, with a promise of eventual success attached, is the false alarm's mirror image and worse - the operator is told to wait for something that will not happen. THE FIX. isDeferrable() classifies, and the two branches say different things. The classification is NOT a new one, which is the point. commitDeferringOnRebalance() already defines what the SYNCHRONOUS path treats as routine - it catches exactly RebalanceInProgressException and CommitFailedException and lets everything else escape - so the async mode now reports at ERROR precisely what the sync mode would have let out, and defers precisely what it defers. Two commit modes disagreeing about which failures are routine would be worse than either answer on its own. RetriableException joins those two because it is the class commitAsync is specified to report transient conditions with (RetriableCommitFailedException extends it) and the sync path never sees it, since commitSync exhausts its own retries first. WHAT IS DELIBERATELY NOT DONE: the review also asked to route permanent failures into the processor's failure lifecycle. Declined, and the javadoc says so where a reader will ask. The async mode has no commit budget and so cannot reach the commit-failure seam (#317, #352) at all - that seam's own options validation REJECTS a commitFailureHandler configured under this mode. Inventing a second escalation policy here is the thing that seam exists to prevent, and it would have to be unpicked when the budget half lands. Throwing is not available either: this runs on the broker-poll thread from a Kafka callback, so an escape strands every waiting committer and takes the instance down - option 1 in commitDeferringOnRebalance()'s own javadoc, rejected there for the same reason. The level is the half that is this PR's to fix. RED PROOF - the new test against the blanket WARN, main code untouched: ConsumerOffsetCommitterAsyncFailureLoggingTest .aPermanentAsyncCommitFailureIsAnErrorBecauseNoLaterRequestCanFixIt a failure no later request can fix must not be reported with the deferral promise / expected to be empty but was: [Async offset commit failed - these partitions stay dirty and are committed when a later request is acknowledged. ...] The captured line IS the false promise, verbatim, which is the whole finding. Tests run: 3, Failures: 1 before; 0 after, and the existing WARN test is unchanged and still asserts no ERROR on a RebalanceInProgressException - so the two tests now pin the level in both directions on both classes of failure. THE #168 BOUND, on the new line too. The ERROR carries its own budget constant rather than raising the shared one: its statement text measures 231 characters against the WARN's 135, and one budget wide enough for both would stop constraining the WARN at all. Both stay orders of magnitude below what the assertion actually guards - DefaultMaxMetadataSize is 4096 PER PARTITION - so a level change cannot become the door the offset map comes back through. The test asserts the bound, the absence of the metadata, and the per-entry rendering on the ERROR line as it already did on the WARN. The solution write-up's failure-line section is rewritten to state the classification, where it came from, and what is deliberately not escalated. Gates: bin/ci-unit-test.sh green across every module; bin/check-all.sh 16 passed, 0 failed; bin/check-issue-refs.sh clean. Targeted run: 45 tests green, including all eleven MockConsumerTestBase subclasses and master's PartitionStateRevocationFenceTest. Upstream-Issue: confluentinc#203 Upstream-PR: confluentinc#345 Forwarded: no Applied-Upstream: no Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019SVm2cT6ZgUyPMim9CukYK
Master gained 6aab3ff, the inflight vetting worklist and the first grooming sweep, which stamped two of the notes this branch resolves. Two conflicts, both in docs/inflight/, and docs/refactoring.md auto-merged cleanly. docs/inflight/bug-async-commit-marked-successful-before-broker-ack.md - modify/delete, resolved by keeping this branch's deletion. Master's edit was a vet stamp asserting the very thing this PR falsifies: "AbstractOffsetCommitter. retrieveOffsetsAndCommit still calls onOffsetCommitSuccess(offsetsToCommit) straight after commitOffsets(...) returns - so an async commit that never lands is marked durable". It does not any more; that is the whole of this branch. A 2026-09-07 stamp saying a defect is still real cannot outlive the commit that removes it, so the note goes, and what outlives it is already migrated to docs/solutions/logic-errors/an-async-commit-was-recorded-on-send-not-on- acknowledgement-2026-09-07.md - the first of the four outcomes in docs/inflight/AGENTS.md, not a bare git rm. This is the same call PR 469's merge made against the same sweep, for the same reason. Nothing else in the tree cited the deleted note. The one reference on master was the "Also surfaced, out of family" bullet in bug-torn-read-family.md, which is the second conflict and is repaired below; a whole-tree grep for the filename and for the note's title now returns nothing, and check-file-refs.sh reports no new dangling reference across 799 citing files. docs/inflight/bug-torn-read-family.md - content conflict, resolved by hand at the markers. Master's sweep shrank the note from the settled candidate dossiers to what is still open, and this branch had rewritten one bullet in the list master also touched. The resolution takes master's rewritten note whole and replaces master's pointer to the now-deleted note - [`bug-async-commit-marked-successful-before-broker-ack.md`](...) with this branch's closed-record bullet, which names what was fixed and cites the solutions write-up. The two bullets git grouped into the HEAD side with it are not this branch's: master deliberately dropped the resetOffsetMapAndRemove- Work bullet and reworded the counter-maps one in its shrink, and both of master's decisions stand. No further sentence needed correcting. Master's shrunk note keeps one claim about the field family this PR changes - "`dirty` can only be set by `onSuccess`, which needs an offset registered by that loop" - and it is still true after the change: setDirty() still has exactly one caller, PartitionState.onSuccess. What this PR moved is the CLEAN mark, not the dirty one; onOffsetCommitSuccess now records the acknowledged offset always and calls setClean() only when the acknowledgement answers the offer the partition last made. Master's 2026-09-08 vet stamp lists what it checked - candidate 2's single-read getShard, candidate 3's single resolution, the concurrent counter maps, the two unmerged racing doubles - and none of those are touched here, so the stamp is left as the accurate record of that reading. Audited in both directions rather than trusting the merge. Against origin/master the result changes exactly the 22 files this branch already changed, and bug-torn-read-family.md differs from master by that one bullet alone. Against dc9a797 it changes 185 files, master's 186 less the note this branch deletes, with no other content of its own. Gates: bin/check-all.sh green - 19 ran, 16 passed, 0 failed, 3 nothing-in-scope, 5 skipped as PR-scoped or argument-scoped. bin/check-file-refs.sh and bin/check-issue-refs.sh green standalone. No JVM run: master changed no Java at all since the merge base, only docs/ markdown and bin/ tooling, so a core unit run would have re-tested this branch's own unchanged commits. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019SVm2cT6ZgUyPMim9CukYK
astubbs
added a commit
that referenced
this pull request
Sep 8, 2026
Master moved two further commits while the previous merge was being verified, both landing in this branch's area: #470 (an async commit counts as committed when the broker answers) and #473 (two quarantine entries cleared by fixing what they were about). Two conflicts, both resolved keeping each side's intent. - `state/PartitionState.java`, `onOffsetCommitSuccess` and `getCommitDataIfDirty`. #470 makes an acknowledgement record the offset always and mark the partition clean only when it answers the offer the partition last made; this branch had the completed-but-uncommitted ledger's `snapshotForCommit()` / `onCommitSuccess()` pair on the same two methods. Both kept. The ledger's snapshot is taken beside the offer, and the trim now sits INSIDE master's `committed.equals(offerLastMadeForCommit)` arm rather than running unconditionally: the snapshot belongs to that offer, so only the answer to that offer may retire what it carried - the same rule #470 introduces, applied to the ledger. This changes nothing in practice, because the ledger is a no-op outside transactional commit mode (`PartitionState`'s constructor gates it on `isUsingTransactionCommitMode`) and the transactional path acknowledges inline with the map it just collected, so that arm is always the one taken where the ledger is real. Said in a comment at the site. - `docs/inflight/test-untracked-ci-flakes.md`. #473 fixed `RegistrationRaceStaleResidentIT`'s setup guard and removed its row and its prose section, adding the "also fixed and out" paragraph; that removal is taken. This branch's `AmbientProbeExtension` row - master's own, with this branch's #225 and 2026-09-03 measurements folded in by the previous merge - is kept, as is its corroboration paragraph in the surviving prose section, and so are the four rows master has never had. Verification: `./mvnw -Pci -pl parallel-consumer-core -am` compiles; 170 unit tests across the producer, transactional, `PartitionState*` and work-manager classes pass with 0 failures and 0 errors from fresh surefire reports - including #470's own `PartitionStateAcknowledgedCommitOffsetTest` and `MockConsumerAsyncCommitCallbackFailsTest`. `bin/check-all.sh`: 16 passed, 0 failed. No Docker-backed test was run. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018rEzrWYFr6oEzy6porczd3
astubbs
added a commit
that referenced
this pull request
Sep 8, 2026
Master landed the async-commit acknowledgement (#470, silent loss on the default commit mode) and the quarantine clearance (#473), which empties docs/quarantined-tests.md so the release guard no longer blocks on it and moves the capacity profiles behind a measured tag. With #466 and #468 already in, tier 1 has four of its seven merged; what is left is #469, #481, #480 and #477. Claude-Session: 460f7df9-dcc2-4b00-a9f9-62f3a2c6d5e4 Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
astubbs
added a commit
that referenced
this pull request
Sep 8, 2026
and #473 Two conflicts. PartitionState: #470 counts an async commit as committed only when the broker answers the latest offer, and #410's ledger trims on commit success; joined so the ledger snapshots before the offer is made and is trimmed only on the answer to that offer - a stale answer to an older offer leaves the newer snapshot in place, so a replay can only over-replay, never under. The flake ledger takes master's updated rows and #473's retirement of the section it fixed, and keeps this branch's three later sightings. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01PVL2FEJ645T76PbybEBUZ6
astubbs
added a commit
that referenced
this pull request
Sep 8, 2026
A SEMANTIC merge, not a textual one. Master gained #470, which changed PartitionState's commit protocol from the other side while this branch was changing it from this one, and git's single content conflict is the smaller half of the work. WHAT 470 CHANGED. Under PERIODIC_CONSUMER_ASYNCHRONOUS the partition used to be marked clean on the line after commitOffsets() returned - that is, at the moment of SENDING, because commitAsync answers later through an OffsetCommitCallback. 470 defers the clean mark to the acknowledgement, which makes two async commits able to be in flight at once, and handles that with one rule inside PartitionState: getCommitDataIfDirty remembers the offer it makes WHOLE, as the OffsetAndMetadata it handed out, and onOffsetCommitSuccess records every acknowledgement's offset through a monotonic recordCommittedOffset but calls setClean() only when the acknowledged object equals that offer. Whole rather than the offset alone, because completing a record above the lowest incomplete one changes the encoded metadata and moves the offset not at all. WHAT THIS BRANCH CHANGED. The dirty flag and stateChangedSinceCommitStart are gone, replaced by a monotone AtomicLong completionCount, a completionCountBeingCommitted sampled once at commit start, and a completionCountCommitted that setClean() publishes. Dirty is isDirtyAt(completionCount.get()) - a comparison of two values rather than a flag somebody clears. HOW THEY COMPOSE. The two rules answer different questions about the same acknowledgement, so they compose rather than overlap: 470's offer decides WHETHER an answer ends the story, and 469's count decides HOW MUCH of it it ends. getCommitDataIfDirty - the only conflicted hunk - now opens the commit window by writing BOTH halves of the pair. This branch's side is kept whole: the count is loaded ONCE into collectedAt, tested with isDirtyAt(collectedAt) and stashed as completionCountBeingCommitted, before the offsets are captured. Master's side is kept whole too: createOffsetAndMetadata()'s result is bound to offered, stored as offerLastMadeForCommit and returned, so the caller and the field hold the same object. Master's reset of stateChangedSinceCommitStart is dropped, because the field it clears no longer exists - the count sampled one statement earlier is what carries that meaning now. onOffsetCommitSuccess is master's, unchanged, with this branch's setClean() underneath it: recordCommittedOffset(committed) always, then setClean() only when committed.equals( offerLastMadeForCommit), else a DEBUG line saying the offset is recorded and the partition stays dirty. setClean() is the count-publishing version, so the value published is the sample taken by the very cycle whose offer was just answered - both fields are written by one invocation of getCommitDataIfDirty, so they can never be crossed. BOTH GUARANTEES SURVIVE, and each has a test that would notice if it did not: 469 - a completion landing inside the commit window leaves the partition dirty, including one landing inside setClean() itself, because setClean publishes a value decided at commit start rather than reading anything (PartitionStateCommitWindowSeamTest, with its control arm replicating the old two-flag protocol and asserting the loss). 470 - an acknowledgement of a superseded offer, including one carrying the SAME offset and different metadata, records its offset and does not clean the partition, and an out-of-order acknowledgement cannot walk the committed offset backwards (PartitionStateAcknowledgedCommitOffsetTest, ConsumerOffsetCommitterOverlappingAsyncCommitTest, MockConsumerAsyncCommitCallbackFailsTest / DroppedTest / SyncCommitMarksCleanOnReturnTest). SENTENCES REPAIRED, each because the merge made it false rather than merely dated: offerLastMadeForCommit's javadoc ended by naming two fields this branch deleted - "a plain field, for the reason {@link #dirty} states" and "the same lifecycle as {@link #stateChangedSinceCommitStart}". Both are dangling javadoc links as well as false claims. The fence is now completionCount's incrementAndGet, and the lifecycle partner is completionCountBeingCommitted; the paragraph now names the pair as it is, says what each half decides, and notes in one parenthesis that the flag it used to name was the count's predecessor. Master's thread-model paragraph and its #466 citation are kept verbatim. completionCountBeingCommitted's javadoc rested on a premise 470 falsified: "AbstractOffsetCommitter #retrieveOffsetsAndCommit runs collect, commit and success callback as one sequential unit". Since 470 the success callback is NOT in that unit under the async mode, so a later cycle can overwrite this field while an earlier request is outstanding. The javadoc now says so, and says why that is safe - setClean() fires only for the LATEST offer, and the count beside it was sampled by that same cycle. Its @ThreadConfined paragraph is restated to match master's model (poll thread under both consumer modes, control thread under transactional, Consumer#close() the one hand-over) and to point at offerLastMadeForCommit as the owner rather than restate it; the volatile and its reasoning stand. onOffsetCommitSuccess's javadoc claimed setClean() "declines when the partition's state changed again while the commit was in flight" - the old check-then-act, which is exactly what this branch removed. It now says setClean publishes the sampled count, so a completion inside the window is not covered by it. getCommitDataIfDirty's javadoc named only the count it remembers; it now names both halves of the window it opens, and says why one is sampled before the offsets are captured and the other after. docs/refactoring.md - both sides edited the AT_STALE_THREAD_WRITE_OF_PRIMITIVE section. Master added a ledger entry describing the commit-window pair as stateChangedSinceCommitStart plus offerLastMadeForCommit, to be @ThreadConfined-declared together; this branch deleted the first of those. Reconciled to one entry naming the pair that exists after both: offerLastMadeForCommit (#470) and completionCountBeingCommitted (#469), with the note that their modifiers already differ deliberately - the offer plain and not to be "fixed" with volatile, the count volatile because no single thread is correct to name across the commit modes. Master's work item, declare the pair together or not at all, is unchanged. docs/inflight/bug-torn-read-family.md - both sides edited it and the two edits are disjoint: this branch's rewording of the dirty sentence and master's replacement of the link to the note 470 retired. It reads true after both; no repair was needed. Two live records master added name setDirty, a method this branch renamed to recordCompletion: docs/inflight/bug-177-commit-response-timeout-unreproduced.md's argument that a stalled instance cannot reach the exception being hunted, and the same argument in CommitResponseTimeoutSoakIT's javadoc. Both are anchors in LIVE documents that must resolve against today's tree, so both are repaired to the current name with the old one in parentheses - the treatment this branch already gave bug-torn-read-family.md. The dated records under docs/plans/ and docs/solutions/ that describe the retired protocol are deliberately left alone: they say what was true when they were written. PartitionStateAcknowledgedCommitOffsetTest's completeAndOffer javadoc explained that getCommitDataIfDirty is what clears stateChangedSinceCommitStart, without which its clean-versus-dirty assertions would be vacuous. The vacuity argument still holds and the mechanism is now the sample of completionCountBeingCommitted; the sentence says that instead. TWO-DIRECTION AUDIT. git diff 7b4691b -- <the three co-edited files> shows master's additions plus the repairs above and nothing else - the lastCommittedOffset javadoc and package getter, offerLastMadeForCommit, the onOffsetCommitSuccess rewrite, recordCommittedOffset, and the offer stash. git diff origin/master -- the same files removes only this branch's own removals - the dirty flag, stateChangedSinceCommitStart, the old setClean/setDirty - plus the repaired sentences. Nothing 470 wrote is lost and nothing 469 wrote is reverted. EVIDENCE. PartitionState*Test, LastCommitTimeFenceTest, ConsumerOffsetCommitter*Test and MockConsumer*Test all green on the merged tree, with both PRs' pinning tests among them: PartitionStateCommitWindowSeamTest 3/3 including its control arm, PartitionStateCrossThreadFieldFenceTest 3/3, PartitionStateAcknowledgedCommitOffsetTest 5/5 including the same-offset-different-metadata case, ConsumerOffsetCommitterOverlappingAsyncCommitTest 3/3, and the three MockConsumer async/sync arms. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_019SVm2cT6ZgUyPMim9CukYK
astubbs
added a commit
that referenced
this pull request
Sep 8, 2026
…-fix Picks up #468, #470, #471 and #473. Read for whether any of them moves the ground under this branch's evidence, since that is exactly what #466 did to the 2026-08-31 grid and is why this branch had to re-calibrate: - #470 is the one that looks like it might. It changes when an ASYNC commit counts as committed - AbstractOffsetCommitter, ConsumerOffsetCommitter and PartitionState - so it is the commit path, but not the revoke path. This branch's grid measures the poll-versus-control lock cycle in tryCommitOffsetsOnRevoke, under the synchronous consumer mode, and the probe's failing edge is the commit-response timeout rather than the clean-marking that #470 rewrites. - #468 (WorkContainer identity equality), #471 (the commit-response timeout soak) and #473 (quarantine entries) touch neither. Checked rather than assumed: the diff of AbstractParallelEoSStreamProcessor.java across this merge contains no change to tryCommitOffsetsOnRevoke, commitLock, onPartitionsRevoked or commitOnRevokeViaTheControlThread - the file is not in the merged range's changed set at all. So the re-calibration stays pinned at 6aab3ff, and the head it names is still the right one to state. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
astubbs
added a commit
that referenced
this pull request
Sep 8, 2026
…, then fenced or redesigned (#469) PartitionState carried three fields that cross the broker-poll / control thread boundary. #349 fenced one of them, dirty, after measuring its window, and deliberately left the other two alone for want of a measurement. This measures them with jcstress and then gives them DIFFERENT fixes, because the measurement says they are different defects. The rule that falls out of it: volatile is the fix for a one-writer field, not for a shared one. THE MEASUREMENT, in jcstress-poc/, in the shape CommitPathVisibilityProbes already used for the dirty pair. Rates are per raced pair on one workstation, not production incidence, and the recorded figures are in each class's Calibration status block. BackPressureFlagVisibilityProbes measures allowedMoreRecords, written on the broker-poll thread inside the commit path's encode and read on the control thread in couldBeTakenAsWork. The plain field with its real surrounding ConcurrentSkipListMap accesses fires; with the neighbours stripped it fires about fifty times more often - so the map touch one statement earlier suppresses the window fiftyfold and does not close it, the same incidental-fencing shape the dirty pair measured; the volatile arm is FORBIDDEN at zero. What the arms do not show, said in the javadoc: this field's harm is unbounded staleness - a stale false admits nothing, so nothing succeeds, nothing re-dirties the partition and the encode never runs again - and "never" is not an outcome a bounded run observes. They measure the fence, not the stall. CommitWindowLostUpdateProbes measures stateChangedSinceCommitStart, written by BOTH threads, with the arbiter asserting the invariant itself: a completion landing inside the commit window must leave the partition dirty unless the commit covered it. As shipped it fires at around 1.6e-3. The arm that decides the design is the flag made volatile as well - the obvious next step, and the one the inflight note had recorded as next: statistically unmoved, because only one of the three mechanisms breaking the invariant is a visibility effect. The check-then-act in the clean-marking step and the lost update between two writers fire at interleaving rates on sequentially consistent hardware, and no modifier addresses either. VolatileStateChangedFlagAcrossTheCommitWindow exists solely to hold that line; re-run it before proposing the modifier again. The monotone-count protocol arm is FORBIDDEN at zero, with a known weakness stated below. THE TWO FIXES. allowedMoreRecords is volatile: one writer, so a release/acquire edge is the whole fix. Its javadoc carries the figures, the arm name and the direction of the crossing - poll thread writes, control thread reads, the OPPOSITE direction to the commit-path fence, so that fence never covered it. dirty and stateChangedSinceCommitStart are gone. In their place a monotone AtomicLong completionCount, the completionCountCommitted a successful commit publishes, and the held completionCountBeingCommitted. "Is this partition dirty" is a comparison of two values rather than a flag somebody clears, which removes the defect rather than narrowing it: no check-then-act is left in the clean-marking step and no write either thread can lose to the other. The release/acquire edge #349 measured is preserved: incrementAndGet is a strictly stronger release than the volatile store it replaces, and the counter load on the commit path is the paired acquire, so the plain longs it published stay plain for the reason that PR gave. The held count is volatile because which thread runs the commit cycle varies by mode, so a thread-confinement claim would have been a lie. COMPOSED WITH #470, which merged first and changed the same protocol from the other side: the partition remembers the commit offer it makes WHOLE and marks itself clean only on the broker's acknowledgement of that offer. The two rules do not overlap. getCommitDataIfDirty samples the completion count once, stashes it, and remembers the offer; onOffsetCommitSuccess records every acknowledged offset and publishes the stashed count as committed only when the acknowledged offer equals the last one made. The offer decides WHETHER an answer ends the commit window; the count decides HOW MUCH it covered. Under the async mode the acknowledgement is no longer part of the collect-and-commit unit, so a later cycle can overwrite both fields while an earlier request is outstanding, which is safe because only the latest offer's answer publishes anything. THE NAMES FOLLOW THE DESIGN. setDirty() is now recordCompletion(): nothing sets a flag any more, the completing thread stamps a new version on the partition, and the count is a version stamp rather than a quantity - its only use is inequality with what the last acknowledged commit covered. isDirty() carries the definition the code had been paraphrasing three ways: dirty means a completion has landed that no acknowledged commit has covered. The comparison lives once, in isDirtyAt(long): isDirty() passes a fresh load, and getCommitDataIfDirty() passes the sample it then stashes, because the collecting cycle must publish the very count it TESTED - a completion landing between two loads would be published as covered while its offset need not be in the metadata this cycle encodes. That burnt commit cycle is the defect this change closes. THE RED PROOF. PartitionStateCommitWindowSeamTest drives, deterministically, the one interleaving the old protocol lost a completion to: inside the clean-marking step, between its read of the flag and its write of dirty. Master offered no override point between those two instructions, so PartitionState.onCommitWindowClosing() was added - an overridable no-op in the one-shot-armed seam shape of RacingSeamWorkManager, with firing tracked separately from arming so a test that forgot to arm cannot pass while asserting nothing. The review searched for an existing hook at that instruction and found none. Redness is shown twice: a permanent control arm reverts only the protocol and asserts the loss, so the defect stays runnable after the fix removed it; and transiently, reverting the protocol in PartitionState while keeping the seam turns the pinning arm red. Stashing the whole change does not produce that, because it removes the seam too. PartitionStateDirtyFlagFenceTest becomes PartitionStateCrossThreadFieldFenceTest, pinning the modifiers on all three fields, which is the move its own failure message asked for. KNOWN WEAKNESS, RULED ON RATHER THAN FIXED. The protocol arm's FORBIDDEN zero is unreachable by construction: the flag arms are seeded dirty so the poll actor always enters the commit window, while the protocol arm starts clean and enters the window only once it has already observed the completion, and the AtomicLong acquire that lets it in publishes everything the release wrote. The recorded run corroborates it - the flag arms reached all four outcomes, the protocol arm only two. Seeding it dirty is not a fix, for the same release/acquire reason; a real power check would need a deliberately broken protocol variant as a negative control. The maintainer ruled that the control would not be built and the weakness documented instead, so its zero is cited as "no anomaly observed" and never as "the window was measured shut". The argument this change rests on is the plain-versus-volatile pair, which none of this affects. Every other FORBIDDEN arm in the module earns its zero from a paired arm of the same state shape that fires; this is the only arm whose fixed version has a different state shape from its comparator. THE PROBE ARMS SHARE THEIR SCAFFOLDING and keep only the measured term apart, answering the duplicate-code check. The rule: a member may be hoisted only if it contains no access to the arm's measured field. The map and its setup, the neighbouring writes, the instrumentation payload, the encode read and the arbiter's result mapping moved into a package-private base class; the measured field's own declaration per @State class and each actor's literal access sequence stay in the arm, so plain versus volatile on one line is the term under test and the JIT compiles the shape production has. jcstress 0.16 permits a @State class to inherit - read out of its annotation processor, whose inheritance ban binds @Result classes only - and jcstress -l lists every arm after the change. The recorded rates are deliberately not overwritten with the re-run's. A LEDGER THAT WAS WRONG, and nothing went red: docs/refactoring.md's AT_STALE_THREAD_WRITE_OF_PRIMITIVE entry listed two fields while SpotBugs named six, three of them in PartitionState - the class #349 was fencing at the time - and one, ProgressTracker.highestRoundCountSeen, that no ledger carried. The lane runs spotbugs:check with failOnError=false, so the finding annotates and never blocks. The repair is the command that re-derives the membership, the spotbugs step's own line verbatim, not a better list. DOCS. The two docs/inflight/bug-* notes for these fields are retired, their durable content in docs/solutions/logic-errors/volatile-is-the-fix-for-a-one-writer-field-not-a-shared-one-2026-09-07.md, and the engine AGENTS.md ledger carries the rule. docs/inflight/test-jcstress-probe-module-open-items.md carries the vacuous-arm line item and the generalisation that a green FORBIDDEN tick is worth nothing unless some arm shows the outcome is reachable at all. Two local-only test failures seen while verifying are recorded in docs/inflight/test-untracked-ci-flakes.md with their same-base control arms; the row for ReactorPCTest.concurrencyTest no longer rules this change out as a contributor, because setAllowedMoreRecords(true) fires on two paths and the fence publishes unblocking sooner too. The master merges kept the retired notes deleted over the vetting sweep's stamps, which asserted the fields were still unfenced and unmeasured - precisely what this change falsifies. Co-authored-by: Claude Fable 5.1 (1M context) <noreply@anthropic.com>
6 tasks done
astubbs
added a commit
that referenced
this pull request
Sep 8, 2026
…d the invariant that keeps it shut is pinned (#484) WHAT WAS OPEN. confluentinc#546 holds three defects behind one warning string. One is fixed by inheritance (confluentinc#563, already in this tree and guarded in OffsetRunLength). One is live. The third - the "Bootstrap polled offset has been reset to an earlier offset" branch of PartitionState#maybeTruncateBelowOrAbove - was never diagnosed upstream, and this fork's note carried an UNTESTED hypothesis for it, proposing a control-arm test nobody had run. This change runs it, and the hypothesis is refuted. THE HYPOTHESIS, AND WHY IT NAMES THE WRONG BRANCH. The note proposed a race: PC issues its own consumer.committed() in loadPartitionStateForAssignment, separate from the consumer's own position resolution for a newly assigned partition, so a commit landing between the two reads would leave PC one commit ahead of where the fetcher starts. It is not a race at all. In kafka-clients 3.9.2 - the version this tree builds on - ClassicKafkaConsumer#updateAssignmentMetadataIfNeeded runs coordinator.poll, which invokes the rebalance listener and therefore PC's committed(), and only afterwards updateFetchPositions, whose ConsumerCoordinator#initWithCommittedOffsetsIfNeeded is what gives an initializing partition its position. PC's read strictly PRECEDES the fetcher's, so a commit landing between the two is the one the FETCHER sees: the polled offset comes out at or above PC's expectation. The proposed mechanism can only produce the other warning. WHAT IS LEFT, AND THE INVARIANT THAT SHUTS IT. One route into the branch belongs to PC rather than to the broker: a commit whose encoded payload decodes to an expectation ABOVE the offset that commit was filed under - the shape #337 closed. PartitionStateBootstrapTruncation162Test round-trips a commit PC itself wrote back through MockConsumer, WorkManager#onPartitionsAssigned and the decoders, over four shapes PC can genuinely commit (fully drained, contiguous tail incomplete, a completed island above the commit frontier, only the base offset incomplete), and asserts the bootstrap expectation equals the committed offset exactly. Hold that invariant and a partition PC hands to itself cannot take either truncation branch. Every remaining way in is the broker handing back a position below the committed offset - an offset reset, a manual rewind, another member committing lower, a stale OffsetFetch across a coordinator failover - and there replaying is the right answer, not a defect. THE TWO BRANCHES AS A CONTROLLED PAIR. One fixture, one differing term - the offset the first poll batch starts at - so the outcome that flips is the branch's own. Below expected: every loaded incomplete is discarded, including one above the batch that the batch does not re-register, and offsetHighestSucceeded resets, so records the previous owner completed and committed are processed again. Above expected, the control: only incompletes beneath the polled offset are pruned, and the one above the batch survives. The note's "duplicate processing, not loss" claim is asserted in BOTH halves - nothing tracked can be stranded below the polled offset, because the expectation the branch fires against IS the lowest tracked incomplete. NEGATIVE CONTROLS, ONE PER ASSERTION, so no arm can pass by accident. Removing the reset fails only the below-expected arm; removing the prune fails only the above-expected arm; dropping the payload's lowest incomplete on decode flips the two round-trip shapes that carry a payload and leaves untouched the two that do not. AN INSTRUMENT THAT LIED FIRST, recorded because it read as a product defect. The round trip initially reported an expectation of 0 for three of the four shapes. MockConsumer#committed hands back a bare OffsetAndMetadata(0) - indistinguishable from a real commit at offset 0 - for any partition its own SubscriptionState does not hold, and MockConsumer#assign clears the committed map, so the assignment has to precede the commit. The arm now proves itself, and the comment at that line says why. A test harness that fabricates a plausible wrong answer is worth writing down: it cost a diagnosis before it was caught. WHAT REMAINS OPEN ON #162, and why this does not close it. The false warning: maybeTruncateBelowOrAbove never asks whether commit data existed, so a partition with no committed offset reports "expected 0 from loaded commit data" and truncates nothing. Misdirection, still untested, and unchanged here. The inflight note is shrunk to that one case and the draft mirror reply is corrected to name it, so the upstream thread is not told that both remaining cases are covered. src/docs/development/upstream-map.yaml records the lifecycle transition in the commit that caused it. CROSS-CHECKED AGAINST THE MASTER THIS LANDS ON. #470 rewrote PartitionState's commit-acknowledgement path while this branch was open, and its offerLastMadeForCommit javadoc names this same downward reset from the other side - only the latest commit offer is kept rather than a set, because the bootstrap reset is "exactly the event no pruning rule would see". The two records agree: the reset is real, reachable only in bootstrapPhase, and destructive of anything keyed to the pre-reset expectation. Nothing in the refutation is disturbed, because the refutation is about how the branch is ENTERED, not what it does once entered. Re-run after merging master rather than assumed. No production code changes: one new test class, the inflight note, and the upstream manifest. Upstream-Issue: confluentinc#546 Co-authored-by: Claude Opus (1M context) <noreply@anthropic.com>
Merged
6 tasks done
astubbs
added a commit
that referenced
this pull request
Sep 9, 2026
…s red control comes first (#491) Closes the COMMIT half of the per-shard liveness gap that docs/inflight/test-per-shard-liveness-has-no-gate.md has owned since the Class 2 lag bound was demoted on 2026-08-25, in the order that note demands: the red control first, the detector only after it. WHAT WAS UNCOVERED. INSTANCE_STALL is per-INSTANCE and is re-armed by any successful work result, so an instance whose other shards keep completing never fires it, however long one partition's commits have been frozen. CLASS2_STALL is the only detector that looks at a partition, and since the demotion it does not gate - because elapsed stagnation is crossed just as readily by a slow record as by a wedge. The correctness ledger does not close it either: it counts records PROCESSED, not offsets durably COMMITTED, so a run balances while leaving exactly the uncommitted offsets a restart would redeliver. THE RED CONTROL, and it is a real engine rather than a script. WedgedPartitionRedControlIT runs one PC over three partitions with a LongPollingMockConsumer that ANSWERS every commit in full and APPLIES only the partitions that are not black-holed - a coordinator that acks and drops, the defect class #470 fixed inside PC. Broker-free, no wall-clock wait on any bound, three arms in one JVM: commits black-holed broker committed 0 PC local next-to-commit 60 FIRES healthy broker committed 60 PC local next-to-commit 60 silent pinned by an broker committed 9 PC local next-to-commit 9 silent incomplete record Every existing gating detector is shown green on the red arm mechanically, not by argument. All 180 records reach the user function, so NO_PROGRESS cannot fire and the ledger balances; the instance completes work throughout and holds nothing at the end, so INSTANCE_STALL re-arms - replayed across twenty times its bound, with an ARMED control that fires at the same spacing so the silence cannot be vacuous. Class 2 observes the frozen watermark and does not gate. THE SIGNAL, and why it is not the demoted bound renamed. UncommittedCompletionDetector compares two POSITIONS, not an elapsed time: a member's own next-offset-to-commit for a partition (offsetHighestSequentialSucceeded + 1, what PC would commit right now) against what the group has actually committed. The Class 2 false positive is excluded structurally rather than by calibration - an incomplete record pins offsetHighestSequentialSucceeded at exactly the offset the broker holds, so the difference is zero however long the stagnation runs. That is the third arm above, and it is why the two replay seeds the demotion was argued from cannot fire it. The consecutive-sample count debounces the COMMIT CADENCE - any commit that lands re-arms it - and the clock does not run while the group is rebalancing. Reachable with no main-code change: WorkManager#getPm, PartitionStateManager#getPartitionState and PartitionState#getOffsetHighestSequentialSucceeded are all public already, so the ShardManager accessor INSTANCE_STALL_BOUND's granularity note declines to add for a probe was not needed. WHAT WAS REFUTED, and it changed the design. The note's own prescription - "a watermark stagnant while completions advance and real backlog exists is the wedge" - does not discriminate as written. Completions advancing is an instance-wide fact, and the false-positive arm has completions advancing, real backlog and a stagnant watermark; a gate built on that wording would have fired on both seeds it was promised to spare. The term that had to change was the granularity of the second signal, from the instance's completions to that partition's own local watermark. WHAT IS STILL OPEN, and the note says so rather than being deleted. The signal reads a PARTITION, and any incomplete offset in it pins the local watermark - so a key-order SHARD that will never be dispatched again inside a partition still gates nothing. Separating that from a merely slow record needs per-shard in-flight state, and no red control exists for it; #483 found the nearest candidate mechanism unreachable, so it is a reachability question before it is a detector one. BOTH REPLAY SEEDS WERE RE-RUN, and they confirm it - the note's second bar. Full chaos suite per seed, on a loaded workstation (load average 19-25, other agents building alongside, which biases TOWARDS crossing a timing bound, not away): seed 6825864417772979246 10 tests 0 failures 0 violations 0 COMMIT_NOT_LANDING 30 Class 2 observations peak lagStagnation 151500ms - OVER the 150s bound peak uncommittedCompletions 3009 records seed 4044221734199516240 10 tests 0 failures 0 violations 0 COMMIT_NOT_LANDING 36 Class 2 observations peak lagStagnation 152020ms - OVER the 150s bound peak uncommittedCompletions 2757 records The lagStagnation peaks are what make this a result rather than an absence: on both seeds the Class 2 bound was genuinely CROSSED, so these are not runs where the false positive simply failed to occur. The new detector was silent through every one. The uncommittedCompletions peaks matter for the opposite reason - a healthy chaos run carries thousands of records of finished-but-not-yet-committed work at its widest, which is exactly why the detector gates on whether the committed offset MOVES rather than on how large the gap gets. ONE INSTRUMENT TRAP, created here and recorded for whoever replays next. The Class 2 interpretation text now names UNCOMMITTED_COMPLETIONS/COMMIT_NOT_LANDING, so a triager meeting a frozen watermark is told what the ABSENCE of that violation means. A log grep for the bare string therefore matches every Class 2 observation: it reads 33 and 39 on these two runs, both of which are zero. Count "COMMIT_NOT_LANDING: partition", or count "VIOLATION:". THE REVIEW'S ONE FINDING, verified before acting on it: localOffsetToCommit re-derives PartitionState#getOffsetToCommit(), which is the definition PC actually commits and could silently drift from the probe's copy. This branch's own two copies collapse into one shared helper naming the method it re-derives. The main-code half - widening getOffsetToCommit() to public, which removes the drift risk outright - is left to the owner as a docs/refactoring.md line rather than taken silently inside the PR that benefits from it, because this suite otherwise declines to widen main-code visibility for a probe. Proven by sabotage: with the detector's reporter never assigned, the red arm fails and both green arms do not. UncommittedCompletionProbeIT pins the re-arm rules a real run cannot reach on demand - a rebalance mid-stretch, a restart, a member that has lost the partition, a commit landing on the last sample before the threshold. The same-class sweep that follows a fix records its one surviving candidate - ProgressProbe#ledger's duplicate allowance is fleet-wide while redelivery is per-partition - and the five detectors it ruled out, in docs/inflight/test-duplicate-allowance-is-fleet-wide-while-redelivery-is-per-partition.md. Co-authored-by: Claude Opus (1M context) <noreply@anthropic.com>
astubbs
added a commit
that referenced
this pull request
Sep 9, 2026
…-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
astubbs
added a commit
that referenced
this pull request
Sep 9, 2026
…producer-fencing-brainstorm Brings the ledger rung's merge of #472, and with it the master that rung merged: the review fix on #472, the commit-window redesign (#469, #470), the revoke commit handed to the control thread (#466), and the grooming sweep (#476). Resolved by hand: - maybeAcquireCommitLock: master's revoke-commit override and this branch's replacement guard compose. A periodic commit waits out a replacement; a revoke commit is answered regardless, since its request was taken for the pass, and while no producer exists it completes with nothing committed - consistent, because the aborted transaction published nothing and no offset was committed, so the next owner reprocesses. controlLoopPass says so at WARN when it happens. - The branch's own assertOnControlThread helper is dropped for master's, which has the same contract. - PCModule.replacementProducerWrap keeps #472's javadoc on why only the replacement build runs as user code, with this rung's sentence that ProducerRecovery builds through it. - TransactionalClaim: master's settled scope note on the revoke-path claim and this branch's C15 recovery claim both stand. - The flake ledger takes master's table, which retired the entries whose tests master fixed or quarantined since, and keeps this branch's one sighting master never had (TransactionTimeoutsTest's long arm). - Two notes the grooming sweep retired are accepted as deleted; the plan's citation of one now points at the history that holds it. - The stack note keeps this branch's heading and impact and takes master's vetting record, attested post-merge-correct. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VrpH51xNDodaajE4P2nhFg
Draft
6 tasks done
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.
Serves #248 (mirror of confluentinc#203) - see What this settles for astubbs/parallel-consumer#248, which is deliberately not a closing keyword.
Description
Under
PERIODIC_CONSUMER_ASYNCHRONOUS- the shipped default commit mode - an offset was treated as committed at the pointcommitAsyncwas sent, not when the broker acknowledged it. A failed or dropped acknowledgement therefore left PC believing an offset was durable when the broker had never recorded it, with nothing dirty left to retry. Silent loss.The trace: what moves, and when
One commit cycle in the async mode, before this change. All of it on the broker-poll thread.
PartitionState.getCommitDataIfDirty()stateChangedSinceCommitStart, so completions arriving during the commit keep the partition dirty. Returns the offset to commitConsumerOffsetCommitter.commitOffsets()AbstractOffsetCommitter.retrieveOffsetsAndCommit()onOffsetCommitSuccess()- the defectPartitionState.onOffsetCommitSuccess()lastCommittedOffsetadvances (thepc.partition.latest.committed.offsetgauge reads it), andsetClean()clears the dirty flagOffsetCommitCallback// todo keep work in limbo until async response is received?Two things the trace rules out, so the diagnosis is not wider than it is: incompletes are not trimmed on commit (
PartitionStatetracks only incomplete offsets and the offset committed is the lowest incomplete one, so there is nothing below it to discard - truncation happens on the bootstrap poll), and there is no commit metric beyond that one gauge. The whole of the damage is two fields, and the load-bearing one is the dirty flag.What a failed or dropped callback left PC believing: that the offset was durable. And because the partition was clean,
collectCommitDataForDirtyPartitions()returned empty from then on, so no furthercommitAsyncwas ever issued for those offsets. Not a delayed retry - no retry. The gauge read an offset the broker never had, and the next owner of the partition resumed from the broker's real position.The fix
Two pieces, minimal and inside the async branch:
AbstractOffsetCommittergainedcommitOffsetsReturnsOnlyOnceAcknowledged(), true by default, and marks success inline only when it holds.onOffsetCommitSuccessbecameprotectedso a committer answering false can call it on its own acknowledgement.ConsumerOffsetCommitterreturnsisSync()from it, and its async branch registersonAsyncCommitAnswered: success records the offsets; a failure logs and leaves them dirty for a later request to carry; a dropped callback does the same by doing nothing at all.A failure is now a deferral, and lands where the synchronous path's rejections already land - logged, not fatal, offsets still marked as needing a commit. That is the option
commitDeferringOnRebalance()'s javadoc argues for at length, reached from the other direction: there, deferral is what a thrown exception causes by aborting before the success marking; here it is what the absence of a success signal causes. Same rule - only an acknowledgement advances the state.The hazard the fix creates, and the single rule for it
Deferring the clean-marking is what makes two async commits able to be in flight at once - before, the first send marked the partition clean, so there was never a second. An answer can therefore arrive for a request a later one has partly overtaken.
The whole of the handling is one rule, inside
PartitionState:Both halves matter, and they are separate because an acknowledgement carries two different things. The offset is always recorded - the answer is true, the broker committed up to it, and
recordCommittedOffsetkeeps the higher of the two when answers arrive out of order, so thepc.partition.latest.committed.offsetgauge cannot walk backwards. The clean mark waits, because marking a partition clean at an offer a later one has passed is what would leave nothing dirty to re-send the offsets in between if that later request then failed or was dropped - this PR's own defect, re-entered through the door it opened.Whole, not the offset alone - and this is the correction the Codex round made to the first version of the rule. The offered offset is one above the highest sequentially succeeded offset, while the metadata riding with it is the encoded set of incomplete offsets above that. Completing a record above the lowest incomplete one therefore changes the metadata and moves the offset not at all, so two requests can be in flight carrying the same offset and different metadata - the ordinary consequence of out-of-order completion, which is the thing this library exists to do. Compared on offsets alone, the answer to the older request matches, cleans the partition, and the newer request's metadata - the only record that the higher record is done - is never re-sent if that request fails or is dropped. Those records are re-delivered after a reassignment: not lost offsets, but exactly the replay the encoded offset map exists to prevent.
Comparing the whole
OffsetAndMetadatais exact rather than merely tighter:commitAsynchands the callback the very map it was given, and the transactional path callsonOffsetCommitSuccessinline with the map it just collected, so an offer's acknowledgement carries that offer's own object. Anything that does not come back identical leaves the partition dirty - the safe edge the rest of the rule already lands on.ConsumerOffsetCommitterkeeps no record of what it has in flight, and that is the point. It passes an acknowledgement straight through, whole. The partition offered the offset, so the partition is the thing that can recognise the answer to its own latest offer.What this revision replaces, and why
On the maintainer's call. Two earlier revisions of this PR put the decision in the committer, and each was a copy of state the partition already held:
asyncCommitSequencethat ignored any answer which was not the latest - throwing away a fact the broker had established, for every partition on the line, including the ones nothing had superseded;highestOffsetInFlightmap, every acknowledgement split into two buckets, and a second success method (onSupersededOffsetCommitSuccess) plumbed throughAbstractOffsetCommitter,WorkManager,PartitionStateManagerandPartitionState- to reach an outcome the partition produces on its own.Deleted with this revision:
highestOffsetInFlight,recordAsInFlight,acknowledgeTheHighestOffsetInFlight,stillCarriesTheHighestOffsetInFlightForAnyPartition, andonSupersededOffsetCommitSuccessin all four classes. Added: onelongfield.commitOffsetsReturnsOnlyOnceAcknowledged()and the sync path are untouched, andsetClean's existing protocol is kept as it is - it still declines when state changed during the commit window, and the counter protocol on #469 is deliberately not pulled in.The per-partition outcome survives, and now falls out instead of being implemented. A commit carries every dirty partition and only the ones that completed more work move, so a request is routinely the newest word on one partition and superseded on another. In the second round the partition that stood still is re-offered at exactly the same offset, so the first round's answer is its latest offer and cleans it, while the partition that moved on stays dirty. A whole-request rule leaves that first partition waiting for a re-commit of an offset the broker has already acknowledged, every cycle, for as long as one partition of an assignment outruns another.
Every edge lands on the dirty side, which costs at most one extra commit and cannot under-report: a partition state rebuilt by a rebalance starts with no offer recorded, so an acknowledgement addressed to the assignment before it cannot clean it either - a change for the better, since the previous code marked clean unconditionally.
The failure line is one line, and its level is the classification
A failed async commit logs one line, and whether it is a WARN or an ERROR is the answer to "can a later request fix this?".
The classification is the synchronous path's, not a second one.
commitDeferringOnRebalance()catches exactlyRebalanceInProgressExceptionandCommitFailedExceptionand lets everything else escape, so the async mode reports at ERROR precisely what the sync mode would have let out.RetriableExceptionjoins those two because it is the classcommitAsyncis specified to report transient conditions with, and the sync path never sees it. Two commit modes disagreeing about which failures are routine would be worse than either answer alone.What this does not do is escalate, deliberately: a permanent failure is logged and the offsets stay dirty, but the instance is not failed. The async mode has no commit budget, so it cannot reach the commit-failure seam (#317, #352) whose job that is - that seam's own options validation rejects a
commitFailureHandlerunder this mode. Throwing is not available either: this runs on the broker-poll thread from a Kafka callback, so an escape strands every waiting committer, which is option 1 incommitDeferringOnRebalance()'s javadoc and rejected there for the same reason.An earlier revision split the level on whether the request was superseded - a distinction only the in-flight map could make, and one nobody on call needed. That is gone; this split is on something an operator can act on.
The #168 (confluentinc#629) bound applies to both levels: the offsets are summarised, never interpolated, the unabridged map stays one level down at DEBUG, and the test asserts the bound on the ERROR line too - with its own budget constant rather than a raised shared one, so a level change cannot become the door the offset map comes back through.
Tests for the rule, and the red proof
The behaviour pins moved to where the rule now lives.
PartitionStateAcknowledgedCommitOffsetTestpins it directly on a realPartitionState: an older acknowledgement after a higher offer records the offset and does not clean; an acknowledgement of the offered offset cleans; a lower acknowledgement after a higher recorded offset moves nothing backwards; and the mixed case, on two partitions, cleans the one it is the latest offer for. Every test reads the offer back rather than naming an offset - the offered offset is derived from the incomplete set, so a hard-coded one would pass against a partition that offered something else.ConsumerOffsetCommitterOverlappingAsyncCommitTestdrives the same three orderings through the committer with two captured callbacks, now over a realWorkManagerand real partition states rather than aWorkManagermock - with the decision insidePartitionState, asserting at the mock boundary would assert almost nothing. It asserts the two things an operator would see: what reached the broker (the capturedcommitAsyncmaps), and what the next cycle would still send (collectCommitDataForDirtyPartitions()- the consequence rather than the flag, and the exact call whose emptiness was the original defect).Red proof, reverting only the
PartitionStaterule - unconditionalsetClean()on success restored, nothing else touched:The older-ack-first case went green-when-it-should-be-dirty in both classes:
{}is the partition marked clean at the older offset while the higher offer is still outstanding, which is exactly the state that leaves nothing to re-send. Restored, all seven green.The out-of-order case passes under both rules - with the clean mark already taken by the newer answer, an unconditional
setClean()on the late one is a no-op. It is kept because it pins the other half: the late answer must not put the partition back into a re-commit of offsets the broker already has.Red proof and the control arms for the original defect
Two red tests landed first, one per way a sent request never becomes a durable commit -
MockConsumerAsyncCommitCallbackFailsTest(the callback arrives with aRetriableCommitFailedException) andMockConsumerAsyncCommitCallbackDroppedTest(no callback at all). The discriminating assertion is the broker's own committed offset, because every weaker one passes on the defect: no exception escapes, PC does not close, and all the records get processed.Both arms flip exactly one term - the new predicate - and nothing else:
commitOffsetsReturnsOnlyOnceAcknowledged()trueexpected to be greater than: 3 but was: 2isSync()falseexpected: 2 but was: 54The third arm is why the green control
MockConsumerSyncCommitMarksCleanOnReturnTestexists, and why it asserts what it does. It checks that commit attempts stand still once the batch is committed - not merely that the offset was reached, which is true either way on the first commit. A sync mode that stopped marking clean would re-commit everycommitIntervalforever, and only the attempt count says so. All four MockConsumer tests are unchanged by this revision and stay green.Why the MockConsumer tests cannot reach the overlap at all
MockConsumer.commitAsyncanswers inline, on the calling thread, before it returns - so a request is always already answered by the time the next one is sent, and two commits are never in flight together. Reaching that state needs the callback held, which needs it captured, which is what the mockedConsumerManagerin the two hand-driven committer tests is for.AsyncCommitterFixtureholds what both need, and now hands back each send's offsets and callback together (AsyncCommitRequest) so a test can say what reached the broker rather than assuming it.One judgement call worth reviewing
ConsumerOffsetCommitterAsyncFailureLoggingTestnow pins the level in both directions: the transient case asserts one WARN and no ERROR, so a false alarm reappearing on a routine coordinator hiccup fails the build, and the permanent case asserts one ERROR carrying the deferral promise's negation. The per-entry term is untouched - that is the half guarding #168 / confluentinc#629, the offset map being interpolated back in atDefaultMaxMetadataSizeper partition. Each level carries its own statement-text budget rather than one raised shared constant, so a level change cannot become the door the offset map comes back through.The Codex review rounds (06f3525, dc9a797)
Two rounds, eight P1 findings. Six applied, two declined with reasoning and ruled on by the maintainer - every one answered in its own thread, all resolved.
Round two raised one finding, on the failure branch treating every exception as a transient deferral. Half of it was this branch's own regression and is fixed in dc9a797 - see the failure line. The other half, routing permanent failures into the processor's failure lifecycle, is declined: the async mode cannot reach the commit-failure seam that owns escalation, by that seam's own design, and throwing from a broker-poll-thread callback is the option
commitDeferringOnRebalance()already rejects.Round one raised the other seven.
docs/refactoring.mdduplicates the field's own thread-model javadocdocs/inflight/issue-response-248.mdis required before merge and is missingdocs/inflight/AGENTS.md; this branch had none where seven siblings existfix(core) astubbs#248: ...Upstream-Issue/Upstream-PR/Forwarded/Applied-Upstreamon 06f3525@ThreadConfinedon the new field and assert itThreadConfined.ANYwould fire on the documentedConsumer#close()callback flush, anddocs/refactoring.mdrecords this field andstateChangedSinceCommitStartas one lifecycle to be declared together or not at allconfluentinc#203Red proof for the metadata finding, new test with the main code untouched, the four existing tests in the class staying green:
Its two guard assertions held first - the offers carry the same offset and different metadata - so it is the real case rather than the ordinary superseded one renamed. The fix is one field changing type:
offsetLastOfferedForCommit(long) becomesofferLastMadeForCommit(OffsetAndMetadata), compared withequals(). The four existing tests now acknowledge by handing the offer object back, which is what the real callback does - reconstructing one from the offset alone tested a request shape that never occurs, and was what let the metadata half go unasserted.Duplicate-code reports
Both flagged clones were between this PR's own files. Three await-and-read blocks were verbatim copies down to the
Collections.singleton(topicPartition)lookup, and are hoisted intoMockConsumerTestBaseasawaitAllRecordsProcessed,awaitBrokerCommittedOffsetandassertParallelConsumerStillRunningWithNoFailureCause.That base class's javadoc carried an explicit decision that assertions are not hoisted, because timeouts are scenario-specific. It is narrowed rather than reversed, and the javadoc records the override where it is made: every helper takes its deadline as an argument, so each scenario still states its own timeout in its own file. The commit-attempt assertion is deliberately not shared, although two of the three files spell it identically - the rejection/withholding families assert it climbing past a floor and the sync control asserts it standing still, which is the distinction those tests exist to draw.
Proved by sabotage that the helpers still bite, reproducing the control-arm numbers already recorded above: forced
true, both async tests red through the hoisted path (expected to be greater than: 3 but was: 2); forcedfalse, the sync control red (expected: 2 but was: 54). All elevenMockConsumerTestBasesubclasses green, including the six untouched.Merged master (7389db5)
#466 landed on
mastermid-review and touchesPartitionStateandPartitionStateManager. One conflict, purely additive - itsfencedForRevocationfield and this PR'slastCommittedOffsetjavadoc were added at the same point - resolved by keeping both, no--ours/--theirs.It strengthens this PR's thread-confinement claim rather than contradicting it: it moves the transactional mode's revocation-time commit off the broker-poll thread and onto the control thread, so
PERIODIC_TRANSACTIONAL_PRODUCERis now uniformly control-thread for both halves of the offer/acknowledge pair. The consumer commit modes keep their inline poll-thread revoke commit, which is the same thread their ordinary commits already use, so both modes stay self-consistent.offerLastMadeForCommit's javadoc cites #466 for this. Its@ThreadConfinedplus runtime assertion onprocessWorkCompleteMailBoxis also the precedent named in the open@ThreadConfinedthread.How this composes with the commit-failure seam (#352)
Verdict, re-checked against #352's own body and its
ConsumerOffsetCommitterdiff: its seam does not cover the async path, and this does not touch what it is building.#352 says so itself, as a deliberate user-directed scoping decision - "Sync consumer and transactional commit modes only. The async mode (the shipped default) has no commit budget and marks offsets clean before the broker answers, so it cannot reach the seam." Its internal record,
docs/inflight/core-async-commit-failure-seam.md(on that branch), names two structural gaps: no budget, so nothing can throw the exhaustion event the handler triggers on; and offsets marked clean optimistically, so the seam's CONTINUE semantics have nothing to attach to.This PR closes the second gap and leaves the first. It adds no budget, no handler and no new option, so #352's options validation - which rejects a
commitFailureHandlerconfigured under the async mode - stays correct on merge: a handler there would still never fire. The two compose by merge order rather than by interaction; theirConsumerOffsetCommitteredits are in different regions (#352 works on the sync waiter, the deferral accounting andCommitResponse; this works on the async branch and the success marking - and this revision leaves that branch smaller than it found it).One follow-up for whoever merges second: that note's "Why the seam cannot reach async today" bullet about optimistic clean-marking will be stale, and the remaining work is the budget half alone.
Other instances of the defect class
The class is state advanced on request rather than on acknowledgement. Checked and dismissed:
ProducerManager.produceMessages/ the produce path - returns the send futures, andParallelEoSStreamProcessorblocks onfutureSend.get(sendTimeout)before the batch is marked succeeded; a failure throws. Correct already.ProducerManager's transactional commit -commitTransaction()is blocking, soAbstractOffsetCommitter's default (true) is right for it and its behaviour is unchanged. It reachesonOffsetCommitSuccessinline with the map it just collected, so the new offer check always matches and the clean mark is unchanged.ConsumerManager.commitSync- blocking, and it already refuses to return normally when close aborted the retry loop, for this exact reason.wc.onUserFunctionSuccess()from the completion signal, andExternalEnginesuppresses the dispatch-time marking explicitly ("a function hasn't succeeded until the inner vertx function has also succeeded"). Same decision, already made correctly.No further instances found.
Prior art
node bin/inflight.mjs prior-art commitAsync OffsetCommitCallback ConsumerOffsetCommitterover 650 refs. What it returned that mattered:docs/inflight/bug-async-commit-marked-successful-before-broker-ack.md- the note this PR retires, which recorded the defect without reproducing or measuring it.docs/inflight/core-async-commit-failure-seam.md(on feat(core) astubbs#317: a commit-failure seam - the application decides instead of PC always terminating #352's branch) - the deferred note naming the two gaps, quoted above.docs/inflight/test-857-churn-storm-async-stalls.md- threePERIODIC_CONSUMER_ASYNCHRONOUSchaos stalls recorded as explained by no known defect: "either a fourth defect or the chaos harness's own teardown races". This is now a defect that operates in that mode, so it is a candidate that ledger did not have. It is explicitly not an attribution - nothing here was measured against those seeds, and that note names the replay as the deciding experiment.docs/solutions/architecture-patterns/two-threads-one-consumer-why-the-commit-seam-keeps-deadlocking.md- which is also where those sightings' mode discipline is recorded.src/docs/development/upstream-map.yaml, entrysweep-2023-broker-disconnect-commit- the 2026-08-07 sweep's verdict on this exact path, quoting the TODO this PR deletes.What this settles for #248
#248 (mirror of confluentinc#203, nioertel 2022-03) reported endless
RetriableCommitFailedExceptionafter a broker restart. Its "Fork status" section names the async callback as the part not addressed, quoting thetodo keep work in limboline verbatim. That part is now addressed.No closing keyword, on purpose, and this is the maintainer's call rather than mine. The reporter's actual symptom was a throughput collapse with endless retries, and nothing here reproduced that - the fix removes the silent-loss path the report exposed, not necessarily the reported slowdown. The issue was closed upstream as completed by an administrative sweep while labelled
wait for info; three diagnostic questions went unanswered. The manifest entry is updated topr-openand says the reproduction attempt with the swept ChaosBroker harness remains the deciding experiment.Documentation
docs/solutions/logic-errors/an-async-commit-was-recorded-on-send-not-on-acknowledgement-2026-09-07.md- the mechanism, the control arms, the seam composition, and the general rule; its hazard section now states the singlePartitionStaterule and records both committer-side designs it replaced, so the next reader does not re-propose one.docs/inflight/bug-async-commit-marked-successful-before-broker-ack.mdis retired, its citation inbug-torn-read-family.mdrepointed at the write-up in the shape that note already uses for its other closed out-of-family finding.docs/refactoring.md- the non-volatile ledger gainsPartitionState's commit-window pair (stateChangedSinceCommitStartand the newoffsetLastOfferedForCommit): same lifecycle, confined by construction, not to be "fixed" withvolatile, and to be declared together if anyone does the@ThreadConfinedwork.src/docs/development/upstream-map.yamlupdated at the lifecycle transition it causes.docs/todo-index.mdregenerated, the deleted marker gone.PartitionState.offerLastMadeForCommit's javadoc owns the rule, including why the offer is held whole, which thread writes and reads it in each commit mode, and why only the latest offer is kept.lastCommittedOffsetcarries the monotonic invariant on the field itself.docs/inflight/issue-response-248.md- the draft response the directory's operator ruling requires before merge. It is deleted when it is posted, not when this PR merges.Gates
Re-run after the Codex round and after merging master:
bin/ci-unit-test.shgreen across every module.bin/check-all.sh: 16 passed, 0 failed.bin/check-issue-refs.shandbin/check-file-refs.shclean.bin/infer-test.shgreen - 10 findings, all known, none new; thestatic: analysislane was RED on the previous head over aNULLPTR_DEREFERENCEratchet identity onPartitionStateManager.lambda$onSupersededOffsetCommitSuccess$4, and this revision deletes the method the finding was in rather than adding a ratchet line for it.Checklist
PartitionStateand committer javadocdocs/features/-N/A - a correctness fix to an existing commit mode, not a feature; no option or API changesPartitionStateAcknowledgedCommitOffsetTestpinning the rule where it lives,ConsumerOffsetCommitterOverlappingAsyncCommitTestdriving the three orderings end-to-end over a realWorkManager, and the failure-logging test now asserting one WARN and no ERROR. Red-proved by reverting only thePartitionStaterule (above)docs/inflight/working note (pr-/branch-) started at the PR's first commit -N/A - the branch's note was the bug note it retires, and everything else here is what gh already showsce-simplifyandce-code-reviewlocally -N/A - reviewed on the PR instead, by both routes that open inline threads. The Claude round's two findings are addressed and the committer-side design they were about has since been removed on the maintainer's call; the Codex round's seven are answered in-thread, five applied and two left open for the maintainer🤖 Generated with Claude Code
https://claude.ai/code/session_019SVm2cT6ZgUyPMim9CukYK