Repository navigation
fix(core): the poll thread never touches the retry queue - the controller collects what it leaves - #481
Merged
astubbs merged 9 commits intoSep 8, 2026
Conversation
…rite lock RED ON PURPOSE at this commit. RetryQueueRebalancePathTest holds RetryQueue's READ lock exactly the way the controller thread does - through a live iterator, which ShardManager.getLowestRetryTime keeps open for a whole scan - and then drives the production rebalance callback on a thread named broker-poll. The lock is constructed fair, so an arriving writer queues behind that scan instead of interleaving with it, and the wait is spent inside consumer.poll() out of max.poll.interval.ms. Five of the six arms fail against this tree: - aRevokeDoesNotTouchTheRetryQueueAndSoCannotWaitForItsWriteLock - the callback is still parked on the write lock when the ten-second deadline passes (ShardManager.removeWorkFromShardFor calls retryQueue.remove). - theStaleSweepDoesNotTouchTheRetryQueueEither - the same, through the epoch-change sweep. Driven by re-assigning an already-assigned partition, which is what increments the epoch and then runs ShardManager.removeStaleContainers in one callback; that method maps the METHOD REFERENCE retryQueue::remove, which is why this reach stayed invisible to ArchitectureTest until #465 taught the walk to follow references. - aDepartedEntryIsCollectedByTheNextControllerPass, aQueueOnlyEntryNoLongerHoldsTheDrainOpen and underOrderedProcessingADepartedEntryIsStillCollectedOnTheSameTick - all assert a controller-side collection that does not exist yet. anEntryWhoseContainerIsStillResidentIsNotCollected is the control and is green here and after: a parked-for-retry container that is still resident is ordinary state, not garbage. The contended-read-lock harness and the parked-for-retry fixture are adapted from #431, which built them for the design this branch supersedes. Its join placement is load-bearing and is kept: the poll thread is joined only after the try-with-resources releases the read lock, because on unfixed code it is parked until then. Reproduce: JAVA_HOME=<jdk17> ./mvnw -Pci -pl parallel-consumer-core -am test -Dtest=RetryQueueRebalancePathTest Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018rEzrWYFr6oEzy6porczd3
…ller collects what it leaves The diagnosis. RetryQueue.remove takes an unbounded, FAIR writeLock().lock(), and every Kafka rebalance callback reached it on the broker-poll thread: by direct call through ShardManager.removeWorkFromShardFor on the revoke and lost paths, and by method reference (.map(retryQueue::remove)) through ShardManager.removeStaleContainers on all three, assigned included. The controller thread holds the matching READ lock for a whole scan - RetryQueue.iterator() hands it out and it is released only at close, which ShardManager.getLowestRetryTime does around an entire iteration - and the lock is fair, so an arriving writer queues behind that scan rather than interleaving. That wait sits inside consumer.poll() with the whole group waiting on it, spent out of max.poll.interval.ms. No cycle was found, so this is an unbounded wait rather than a deadlock. What changed for a user. A rebalance can no longer be delayed by however long the control loop's retry-queue scan takes, in any ordering mode or commit mode. The fix. The callbacks remove from the SHARDS only - a ConcurrentHashMap of ConcurrentSkipListMaps, with no lock to wait for and so nothing to decline. ShardManager.purgeDepartedRetryEntries() then collects, on the controller thread where waiting is allowed, every retry-queue entry whose container ProcessingShard.isResident says has left: it scans under the read lock, closes the iterator, and removes the collected entries under the write lock in one acquisition via removeAll. The invariant, stated once at the purge: a retry-queue entry with no resident container is garbage the controller collects; it may exist for at most one control-loop tick; the callbacks are free to create it. Where it runs, and why there. At the top of ShardManager.getWorkIfAvailable - the one point the control loop reaches unconditionally every pass through retrieveAndDistributeNewWork in both RUNNING and DRAINING, and before drain() reads getNumberOfWorkQueuedInShardsAwaitingSelection in that same pass. The drain path is the consequence that matters: #437 established that a queue-only orphan survives that figure's flooring once the pipeline is drained, so it reads true forever and a draining close hangs to its timeout. The bound is one control-loop tick in every ordering mode. The purge scans the retry queue rather than the shards, so the ordered modes' shard-scan break - which stops the scan at the first container handed out - cannot delay it. That is the one place this design is strictly better than the rejected one rather than merely smaller. Why the scan and the removal can be two steps. Departure is monotonic (isResident is reference identity and nothing re-inserts the same instance), and nothing else writes the queue's keys: RetryQueue.add is @ControllerThreadOnly, its only production caller is ShardManager.onFailure, and the purge runs on that same thread. The second half matters because RetryQueue removes by topic/partition/offset and so cannot say which container it meant - the defect class of #468. What would reopen it is a second writer. The rejected design, named. #431 kept the poll thread on the queue and made it DECLINE: tryRemove, abandon the paired shard removal on refusal, and after #437 a second ask after the shard removal with a put-back on refusal. It is correct and it is proven; it costs a second non-blocking entry point keyed differently from the blocking one, a queue reference threaded into a second ProcessingShard method, two asks per removal, and a retirement bound that varied by ordering mode. Over the same three main files it is +347 -72 against +185 -67 here, most of the latter javadoc. Also rejected: reusing #466's control-thread hand-off, which has the poll thread WAIT on a CompletableFuture - right for a revocation commit, wrong here, because the purge needs no hand-off at all and reusing that shape would reintroduce the wait being removed. Counters during the window, re-measured on this tree. Between a callback's shard removal and the purge, getWorkableRecords (inShards - parkedForRetry) reads one LOW - negative in the single-record case. Low is the safe direction: isSufficientlyLoaded() is workable > threshold, so it fetches sooner; high would pause the poller with nothing to process. WorkManagerTest's revocation arm was re-stated to pin the direction and that one pass settles it. #437's post-add residency confirmation in ShardManager.onFailure becomes belt-and-braces - the purge would collect the orphan it prevents a tick later - and is kept this round, with its javadoc saying so and naming it as a live question rather than a settled one. Evidence. Red-then-green: the five red arms of the previous commit are green, with the control arm green throughout. Ablations: deleting the purge call turns six arms red across three classes; inverting the residency predicate turns seven red - the same six plus the control. The ArchUnit rule was proven on merit rather than assumed: with KNOWN_BLOCKING_VIOLATIONS emptied and the callbacks still reaching the queue it reports 24 violations (18 exemption keys; the @ControllerThreadOnly half fires once per route), and green afterwards with the set still empty. All eighteen entries are deleted; the set, its root => target keying and the argument for keeping it remain, because an exemption is how a newly-found reach is recorded as a defect on the books. Also: RetryQueue.removeAll widens to Collection<? extends WorkContainer<?, ?>> - the body only reads topic/partition/offset, and no <K, V> signature accepts the WorkContainer<?, ?> the queue's own iterator yields without an unchecked cast. Error Prone's UnnecessaryTypeArgument found the three test call sites that carried a now-redundant type witness. Two existing arms are re-stated rather than deleted, because the sweep they described stopped existing: theProductionSweepsQueueRemovalIsGatedOnItsShardRemoval becomes theProductionSweepLeavesTheRetryQueueToTheController (the sweep touches no queue; the controller pass then collects exactly the departed entry and leaves the resident one), and theSameRebalanceOneSeamLaterTakesTheQueueEntryWithTheShardEntry becomes ...LeavesTheQueueEntryForThe ControllerToCollect. aQueueFirstSweepDefeatsTheOneShotConfirmation keeps its name and assertion - two write-ups cite it - and gains a javadoc saying it now models a superseded design rather than an incoming one. Collides textually with #468, #431, #410, #392 and #362 on the sweep, and #408 on ArchitectureTest; none of them conflicts in design. Reproduce: JAVA_HOME=<jdk17> ./mvnw -Pci -pl parallel-consumer-core -am test \ -Dtest='RetryQueueRebalancePathTest,RetryQueueRequeueWindowTest,WorkManagerTest,ArchitectureTest' Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018rEzrWYFr6oEzy6porczd3
…ve records that named #431 The defect is fixed, so its in-flight note has stopped describing reality. Following the four outcomes in docs/inflight/AGENTS.md, what outlives the work is migrated first and only then is the file removed. New: docs/solutions/runtime-errors/retry-queue-write-lock-on-the-rebalance-path.md records BOTH designs. Design A (#431) prevents the split - decline with tryRemove, abandon the paired shard removal on refusal, ask again after it - with why it was correct, why it was proven, and why it lost on machinery rather than on correctness. Design B is what shipped: the poll thread never asks, and the controller collects retry-queue entries whose container is resident in no shard. It carries the invariant, the one-tick bound, the two-step scan's safety argument, the counter direction, the rejected alternatives (including reusing #466's hand-off) and the red-then-green evidence. The write-up is adapted from the one Design A wrote for itself, so its diagnosis sections are preserved rather than re-derived. Removed: docs/inflight/bug-retry-queue-write-lock-on-the-rebalance-path.md (retired into the above) and docs/inflight/pr-431-must-pair-its-queue-removal-with-the-shard-removal.md, whose whole subject - #431's queue-first sweep needing a second removal - is a property of a design that will not land; its reasoning is in Design A's section. Added: docs/inflight/branch-retry-queue-controller-purge.md, for the three things no command will tell a reader - that #431 is still open and must not be merged, which open PRs collide textually on the sweep, and which decisions this deliberately leaves open. Corrected beside their dated text, never over it, because each is a record of a measurement or a state at a date: - static-archunit-main-code-rules.md - KNOWN_BLOCKING_VIOLATIONS is now empty on merit, measured by deleting it with the reaches still in place (24 violations). The paragraphs describing the list as it was stand; what is corrected is that its named owner is superseded, and that the tryLock sibling which would have made @ControllerThreadOnly the only discriminator does not exist. Its citation of the retired note is repaired in place as a history pointer, which is the repair docs/citations.md prescribes and the file-ref gate recognises. - test-retry-queue-behaviour-untested.md - two of its four items are now covered, by removal rather than by symmetry; the other two are still open and now have RetryQueueTest and RetryQueueLincheckTest to read first. - bug-shard-displacement-orphans-the-retry-queue-entry.md - the displaced container's entry is now collected on the next pass, so the FIGURE it names as the harm is bounded by one tick; the pairing gap itself and its production-reachability question are unchanged. - core-retry-queue-needs-a-runtime-controller-ownership-guard.md - its precedent is superseded, and the purge's safety resting on the queue having exactly one writer makes that guard more load-bearing, not less. - test-largenumberofinstances-residual-failures-measured-not-explained.md and static-infer-threadsafe-is-blocked-by-third-party-interfaces.md - both point a reader at #431 by number; the refutation in the first is untouched and explicitly not to be re-run. - docs/solutions/runtime-errors/retry-queue-orphan-window-between-the-requeue-check-and-the-add.md gains a dated correction: #431 will not land, so the assertion it says to invert stays as written, the confirmation it is about is now belt-and-braces, and "what would reopen it" is superseded in one direction and sharpened in the other. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018rEzrWYFr6oEzy6porczd3
That sweep stamped or restated nearly every in-flight note, including all seven this branch touches, so the conflicts are all in docs and all about the same fact: it recorded #431 as "the fix in flight", which this branch supersedes. Resolved: - bug-retry-queue-write-lock-on-the-rebalance-path.md and pr-431-must-pair-its-queue-removal-with-the-shard-removal.md - modify/delete, resolved as DELETE. Master's additions to both are a vetting stamp confirming the defect was intact on 2026-09-07/08 and a rewrite of "Fixing it" naming #431. Neither outlives the fix: the stamp's finding is the problem this branch closes, and the design it names is the one this branch rejects. Both designs and the evidence are in docs/solutions/runtime-errors/retry-queue-write-lock-on-the-rebalance-path.md. - test-retry-queue-behaviour-untested.md - content conflict, resolved by TAKING MASTER'S SHRUNK NOTE and re-stating this branch's correction against it. The sweep had already closed the two items the correction addressed, and better than the correction did; what it could not know is that two sentences in its own text name a mechanism this branch removes - `removeWorkFromShardFor`'s ordering, which no longer exists because that method no longer touches the retry queue. The correction is now only about those two sentences. Auto-merged and checked rather than trusted: bug-shard-displacement-orphans-the-retry-queue-entry.md, static-infer-threadsafe-is-blocked-by-third-party-interfaces.md and test-largenumberofinstances-residual-failures-measured-not-explained.md each gained a dated vet stamp above this branch's dated correction, which is the intended shape - the stamp records what was true on its date, the correction says what moved on 2026-09-08. Two stamps make claims this branch falsifies (#431 still open as the reason a clause has not gone stale; RetryQueueRebalancePathTest existing only on that branch) and are deliberately left standing, dated, with the correction beneath. No vet markers are added by this branch: docs/inflight/AGENTS.md's "Vetting a note" writes the marker only on the outcome where a note is still true AND unchanged, and every note here is shrunk, corrected or removed - which is its own record. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018rEzrWYFr6oEzy6porczd3
… and one stale comment Three reviewers (reuse, quality, efficiency) over this branch's diff. Two findings applied, one recorded rather than applied, and the reason for that is the interesting part. APPLIED - the new test hand-rolled a harness BlockedThreadAsserter already is. It built its own thread, CountDownLatch, bounded await and join, which is exactly that class's machinery; the repo's rule is to extend the shared helper rather than grow a parallel copy, and a drifted copy of test setup has already been a flaky-CI source here. BlockedThreadAsserter gains a third idiom, `returnsWithin(function, threadName, budget)`, and `joinQuietly`. It is the NEGATIVE of the two it had - they assert a function blocks, this hands back whether one returned - and a caller proving a non-block usually cannot give up its own thread to an assertion helper, because it has to hold the thing the function would wait for and read state while the contention is on. That is why it returns the fact rather than asserting. What the reuse buys beyond deduplication: `runCapturingThrowable`. A callback that throws on entry returned promptly, so the hand-rolled version could not tell it from a callback that did its job - a false PASS for a test asserting promptness. `returnsWithin` raises it instead. The thread name is the caller's, because these tests are about which thread a call is on and `broker-poll` is what a stack dump has to say. Re-proved after the refactor rather than assumed: restoring the queue removal to both callback paths turns three arms red again, the two lock ones by timing out at the deadline. BlockedThreadAsserterTest (the negative control for the two existing idioms) and ProducerManagerTest (its main caller) are green. APPLIED - `ShardManager.onFailure`'s residency branch still described a race that no longer exists. Its comment and log said a `false` from `retryQueue.remove(wc)` means "the revoke sweep's own paired removal got there first", which was true when the sweep removed from the queue. It cannot happen now: the add is a few lines above, the sweep never touches this queue, and every other removal is on this same controller thread later in the pass. So TRUE is the only outcome production can produce, and FALSE became a SIGNAL - it would mean a second writer of the retry queue exists, which is the one thing that invalidates purgeDepartedRetryEntries()'s scan-then-remove. Said that way at the site, since nothing asserts it. RECORDED, NOT APPLIED - both the quality and the efficiency reviewer independently found that the purge is the first UNCONDITIONAL full scan of the retry queue on the control loop. Every other reader there early-stops on the retryDueAt sort order; shard residency has no relationship to that order, so this one cannot, and it finds nothing on the overwhelming majority of ticks. The fix they converge on is to gate the scan on a departure signal - and the review sharpened what that signal should be: a monotonic LongAdder in the shape RecordPopulation already uses at those very call sites, not the boolean flag the design rejected, which also keeps the one-tick bound trivially. It is not applied here because it puts poll-thread-written state back on the rebalance path, in the change whose whole point is taking the poll thread off this structure - and because the cost is unmeasured. DispatchScanMeter is this class's own precedent for metering exactly this, and the honest order is meter, then decide. Recorded in the write-up's rejected alternatives and in the branch note, with that order stated, so it is a queued decision rather than a lost review comment. Also from the same pass, and deliberately NOT acted on: the reviewers confirmed the ArrayList allocation on an empty tick is free (the no-arg constructor defers the backing array), and that the fair read lock held across the scan cannot hurt the poll thread any more, because the poll thread no longer touches this lock at all. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018rEzrWYFr6oEzy6porczd3
Owner
Author
|
@claude review this |
|
Claude finished @astubbs's task in 4m 7s —— View job Review
Note on tooling: my allowlist here comes from SummaryThis is a clean, well-reasoned fix. Moving the rebalance callbacks off
Test coverage
The Minor observations (non-blocking)
VerdictThe design decision (no declining alternative needed, since the callbacks never need to ask) is sound, the invariant it depends on is correctly proven from the actual code rather than merely asserted in prose, and the test suite exercises the concurrency property directly rather than through timing coincidence. I'd merge this over #431 as described in the PR body — nothing here gave me a reason to object to that framing. |
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)
No new clones introduced by this PR. Powered by astubbs/duplicate-code-cross-check |
[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. |
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #481 +/- ##
============================================
+ Coverage 82.31% 82.85% +0.53%
- Complexity 1538 1560 +22
============================================
Files 96 96
Lines 5343 5377 +34
Branches 529 534 +5
============================================
+ Hits 4398 4455 +57
+ Misses 756 729 -27
- Partials 189 193 +4
Flags with carried forward coverage won't be shown. Click here to find out more. ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
|
🟢 Throughput — OKThis branch measured about 6% 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 1 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: ad482ec Since the previous push: ratio 1.115 -> 1.058, share 1.435 -> 1.804, rate 79514 -> 85118 (+7.0%). One push of difference sits inside this test's measured spread - read it as movement, not as a result. Updated for |
Owner
Author
|
@codex review |
|
You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard. |
astubbs
marked this pull request as ready for review
September 8, 2026 02:02
…es introduced `bin/check-pr-analysis-surfaces.sh 481` reports findings on lines the diff WROTE separately from ones inherited in files it touches. Three of the former, all DLS_DEAD_LOCAL_STORE, all from the same habit: naming a discarded return value. The repo has already settled this tension and the resolution was two feet away, at `ShardManager.onFailure`: "a dead store in main code trades one static-analysis finding for another". The root AGENTS.md rule - never discard a return value silently, name it and say why - is about a reader being unable to tell a decision from an oversight; LOGGING the value satisfies that better than an ignored local, and satisfies the analyser too. So each site now uses its value rather than parking it: - `removeWorkFromShardFor` logs what the shard gave up at trace, beside the trace its else-branch already writes. The line says the retry queue is deliberately untouched here, which is the fact a reader arrives at this method wondering about. - `purgeDepartedRetryEntries` returns void. Nothing in production read the count, and a caller that named one only to drop it is exactly the dead store; what is worth knowing it already logs. Said in its javadoc, along with where a meter would go if the scan's cost is ever measured - the deferred optimisation both reviewers asked for. - The purge's `removeAll` answer is folded into the log it already writes when something was collected, and it is the same signal as the one at `onFailure`: this thread is the queue's only writer, so `false` can only mean a second writer removed them first - which is precisely what would invalidate the scan-then-remove the purge is built on. Inherited findings in the same files are left alone, as that script's own output instructs: they belong to the registries under docs/inflight/. Reproduce: bin/check-pr-analysis-surfaces.sh 481 Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018rEzrWYFr6oEzy6porczd3
6 tasks done
Owner
Author
|
Predecessor: #431, now closed as superseded by this PR. Its contended-read-lock harness and parked-for-retry fixture are what this branch's red test adapts, and its design is Design A in the solutions write-up here. |
…nse what waited on it The owner closed #431 as superseded. Six places on this branch said it was open; each is now correct, and two of them were making a claim that mattered rather than merely naming a number. Retired: docs/inflight/branch-retry-queue-controller-purge.md. Its whole first purpose - "#431 is still OPEN and must not be merged" - is fulfilled, and the four outcomes in docs/inflight/AGENTS.md put what remains where it is owned rather than deleting it wholesale: - the collision list is answerable by `gh pr list` plus `gh pr diff --name-only`, and is in this PR's body where a reader of the change will meet it; - the belt-and-braces confirmation, the runtime ownership guard and the shard-displacement orphan are each already owned by ShardManager.onFailure's javadoc, the solutions write-up, and their own notes, all corrected in this branch's earlier commits; - the one item with no owner was a SPLIT, not a deletion, and is now docs/inflight/perf-retry-queue-purge-scan-is-unmetered.md - the purge's unconditional full scan, the LongAdder-gated shape both reviewers converged on, why it was not taken with the change, and the order that would settle it (meter first, using DispatchScanMeter's precedent, then decide). It carries `deferred` with that as its reason, so it sorts as a decision waiting on evidence rather than as work nobody did. Past-tensed beside the dated text, never over it: - ShardManagerLincheckTest's fixture javadoc rested on "would go red the moment that PR lands". That event will never happen, so the urgency is void - and the fixture change was still right, for the half of its argument that never depended on the PR: the reach is created by a state the product deliberately makes, and a harness whose fixture contradicts its own comment is a false green whatever the schedule. Said there, because the next reader of that comment is the one who would otherwise take the void premise for a live one. - core-stale-arrival-guard-needs-a-null-safety-decision.md carried the same premise for the same sighting. Its datum is explicitly unaffected - it is about that note's own question, not about the PR's outcome - which is why the paragraph stands with a correction rather than being cut. - The two 2026-09-07 `inflight-vetted` stamps that read "#431 is still OPEN" are left exactly as written - they were true on their date, and that is the point of a dated stamp. What each note's 2026-09-08 section now says is that the stamp's REASONING no longer holds: the shard-displacement clause is dead for a different reason than the stamp gives, and largeNumberOfInstances' "false positive because RetryQueueRebalancePathTest lives on #431" resolves because that test now lives here. - The orphan-window write-up's correction is sharpened from "superseded, not merged" to "CLOSED, never merged", with the consequence stated: the arms it promises to invert stay as written. Gates: check-branch-self-reference (now with PR #481 resolved, so the number phrases are checked too), check-file-refs, check-inflight-tags, check-issue-refs, and the full bin/check-all.sh - all green. ShardManagerLincheckTest re-run green after its javadoc edit. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018rEzrWYFr6oEzy6porczd3
Owner
Author
|
@claude review this |
|
Claude finished @astubbs's task in 1m 38s —— View job Review
Note on tooling: same limitation as the prior automated review — my allowlist is SummaryThis is a re-review after three more commits landed on top of the state the first automated review (01:42 UTC) already approved: The three new commits, checked individually
VerdictNothing in the new commits changes the design or introduces a new concern beyond what the first review already covered and endorsed. The dead-store fix is exactly the kind of narrow, low-risk cleanup it claims to be, the simplify pass is a genuine quality improvement with re-verification stated, and the docs commit touches no code semantics. I have no new blocking findings; my read is unchanged from the earlier review — I'd merge this over #431 as the PR describes. |
astubbs
added a commit
that referenced
this pull request
Sep 8, 2026
…rphan is known unreachable Two things moved under the burn-down on 2026-09-08. The owner's #481 supersedes #431 for the retry-queue lock on the rebalance path: the poll thread never touches the queue and the controller collects what it leaves. #431 is closed; tier 1 now lists #481, ready and green, and notes that #483 stacks on it. The sweep's reading, moved into this note earlier, still called #431 a draft; it is annotated rather than rewritten, since it is the sweep's dated reading. The shard-displacement orphan window is known unreachable by #483, with the one caveat worth carrying: the proof's last leg is a Kafka property (the fetch position never goes backwards within a generation), so an in-generation offset replay reopens the window and nothing goes red. #481's purge bounds the cost to one tick of misdirection. Two more by-key removals of the same shape are named and left for #468, whose identity-equals change is what makes conditional removal possible. Claude-Session: 460f7df9-dcc2-4b00-a9f9-62f3a2c6d5e4 Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
[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
added a commit
that referenced
this pull request
Sep 8, 2026
…te leaves open #481 takes the rebalance callbacks off the retry queue and has the controller purge entries whose container has left. It edits the same three files this work does, and its "Update 2026-09-08" section on the shard displacement note ends by saying production reachability is still not established - which is exactly what this work answers. Merged so the two read as one argument rather than two adjacent ones. WHAT CONFLICTED, AND HOW IT WAS RESOLVED ONE textual conflict, in ShardManager.onFailure's javadoc. Both sides rewrote the same trailing paragraph about addWorkContainer. #481 said the purge now backstops the displacement orphan; this branch said the branch is unreachable. Both are true and neither implies the other, so #481's paragraph is kept verbatim and the verdict is layered underneath it: the purge bounds the harm whatever happens, the proof says the case does not arise, and the backstop is therefore genuine belt-and-braces rather than the only thing standing between that branch and an orphan. TWO auto-merges that were WRONG and needed hand repair - the reason a clean auto-merge was not taken as a resolved merge: - ProcessingShard.addWorkContainer. Both blocks survived, but the cleared- suspicion text asserted that each staleness transition is "followed by the sweep that removes it from BOTH structures". #481 makes that false - the callbacks remove from the shards only. Rewritten to say what the argument actually needs, which is the SHARD removal: a container the sweep unseated is not a resident, and the displacement branch needs a resident to displace. The queue half was never the load-bearing part. - The inflight note. Collapsing the two vet markers to one (the tag gate permits exactly one) deleted the 2026-09-07 stamp that #481's update paragraph refers to by date. That paragraph now records what the stamp said and that it has been superseded, so its point survives its source. WHAT CHANGED IN THE WORK ITSELF - Leg five of the proof is restated in terms of residence, in the solutions write-up, the test's class javadoc and the code comment. - Arm one is renamed aRebalanceUnseatsTheResidentBeforeAnyReplacementCanArrive and now asserts the two removals in #481's order: gone from the shard inside the callback, gone from the queue after one controller pass. It asserts the one-tick bound rather than assuming the old pairing. - Arm three additionally shows the orphan it constructs does not survive one control-loop pass. That assertion deliberately does NOT attribute the removal to the purge: under a fence the last-resort stale sweep would take the same key, and this arm cannot tell them apart. A first attempt asserted the replacement would be handed out on that pass; it was refuted on the run - the fence makes the whole partition stale, so the replacement is stale on arrival too. The assertion now records that. EVIDENCE AFTER THE MERGE ShardDisplacementOrphanReachabilityTest 3/3, #481's RetryQueueRebalancePathTest 6/6, RetryQueueRequeueWindowTest 9/9, ProcessingShardStaleReplacement909Test green. bin/check-all.sh 16 passed, 0 failed. The ablation was RE-RUN on the merged tree rather than carried over, because #481 changes what those sweeps do and an ablation measured against the old shape would be evidence about code that no longer exists. Same result: either sweep alone green, both removed red at the load-bearing assertion. Co-authored-by: Claude Opus <noreply@anthropic.com>
astubbs
added a commit
that referenced
this pull request
Sep 8, 2026
…rom the other side #468 merged to master (7c95b75) between this branch's #481 merge and now. It is the defect class this work's sweep reports on, so leaving it unmerged would have published a sweep naming an open PR as the owner of candidates that PR had already settled. WHAT CONFLICTED ONE conflict, in ProcessingShard.getWorkIfAvailable's last-resort stale sweep, where both sides rewrote the same comment. Neither side was a superset: - #481 replaced the "orphans it forever / the load gate reads low" claim, which its purge makes false, with the purge that now collects the entry. - #468 added why the removal is conditional on the inspected container, and a dated CLEARED SUSPICION for this site. Resolved by keeping BOTH - #481's purge sentence in place of the false claim, and #468's conditional-removal reasoning and cleared suspicion after it. Taking either whole side would have silently dropped the other's content, and a green build would not have shown it. THE RESULT WORTH RECORDING #468's cleared suspicion at that sweep rests on exactly the fact leg one of this work's proof rests on: addWorkContainer is the ONLY writer of workMap outside tests and runs on the controller. Two people reached the same single-writer discriminator from opposite directions, days apart, neither citing the other. That means the two clearances share ONE reopening condition - anything that puts into a shard off the controller thread - and nothing gates it. Said at all four places that now depend on it: both code comments, the solutions write-up, and the test's class javadoc, so a future change cannot invalidate one while reading only the other. CITATIONS BROKEN BY THE COMBINATION, AND REPAIRED #468 retired docs/inflight/bug-stale-sweep-iterator-evicts-fresh-replacement.md when it merged; #481 cites that note in ShardManager.onFailure's javadoc and in its solutions write-up. Neither PR is wrong on its own and neither would have failed its own gate - the dangling reference exists only in the merge. Both repaired the way docs/citations.md and check-file-refs.sh prescribe, by pointing at the history that still holds the file (git show 7c95b75^:docs/inflight/bug-stale-sweep-iterator-evicts-fresh-replacement.md). EVIDENCE ShardDisplacementOrphanReachabilityTest 3/3, RetryQueueRebalancePathTest 6/6, RetryQueueRequeueWindowTest 9/9, ProcessingShardStaleReplacement909Test 4/4, ShardManagerStaleContainerTest and ShardAvailableCountOwnershipTest green - 31 tests, 0 failures. bin/check-all.sh 16 passed, 0 failed. The ablation was re-run a THIRD time, on this tree. #468 changes each sweep removal into a compare-and-remove, so the previous run was again evidence about a shape that no longer exists. Same result: either sweep alone green, both removed red at the same assertion. The write-up now records all three runs rather than the latest one. Co-authored-by: Claude Opus <noreply@anthropic.com>
6 tasks done
…ntries Picks up #466 (revoke-path commit hands itself to the control thread), #468 (WorkContainer equality is identity, so the stale sweep removes only the container it inspected), #471 and #476. One conflict, comment-only, in ProcessingShard.getWorkIfAvailable's controller-side stale branch: #468's conditional-removal reasoning and its dated cleared suspicion are kept; its sentence "leaving the queue entry behind orphans it forever - nothing else removes an entry whose container is in no shard" is replaced by this branch's, because ShardManager.purgeDepartedRetryEntries() now does exactly that on the pass after. The code beneath merged clean: #468's evictIfStillResident plus the controller-thread retryQueue.remove. Two citations of docs/inflight/bug-stale-sweep-iterator-evicts-fresh-replacement.md - which #468 retired - repaired to history pointers per docs/citations.md, in the solutions write-up and ShardManager's javadoc. #468 also settles the one premise this branch's purge rested on by argument: it removes entries by key while equality is identity, safe here only because the queue has one writer. Verified on the merged tree from fresh reports: RetryQueueRebalancePathTest, RetryQueueRequeueWindowTest, RetryQueueTest, RetryQueueIteratorConfinementTest, ShardManagerTest, ShardPopulationRaceTest, WorkManagerTest, WorkManagerStaleCheckDoubleLookupTest, ArchitectureTest, RebalanceCallbackRuleControlTest - all green. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018rEzrWYFr6oEzy6porczd3
[superseded - the quarantine lane is now empty] 🧪🔒 Quarantine Lane ReportThe superseded report, collapsed because it no longer applies
🔴 expected while the owner PR is open · 🟡🎲 flapper, pass proves nothing · 🚨 a deterministic quarantined test passing means its fix landed: delete its Since the previous push: Updated for Superseded by a newer quarantine lane report. |
…ntries One conflict: the residual-failures note this branch had appended a dated update to was deleted by #473, which migrated it to docs/solutions/test-flakiness/. The note stays deleted; the update - that #431 closed as superseded by this branch and the refutation stands unchanged - follows the content as a citation note beside the #431 reference in the write-up, per docs/citations.md. Claude-Session: 460f7df9-dcc2-4b00-a9f9-62f3a2c6d5e4 Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
🧪🔒 Quarantine Lane ReportThe quarantine lane is empty - no Any earlier row on this PR asking for a Lane: non-gating; rules: see the Quarantine Audit check. Since the previous push: Updated for |
astubbs
added a commit
that referenced
this pull request
Sep 8, 2026
One conflict, inherited from the #481 tip this branch carries: the residual-failures note #481 had appended to was deleted by #473, which migrated it to docs/solutions/test-flakiness/. The note stays deleted here; the dated update that followed it lives on #481 as a citation note in the write-up and arrives with that PR. 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
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
deleted the
fix/retry-queue-controller-purges-departed-entries
branch
September 8, 2026 03:27
astubbs
added a commit
that referenced
this pull request
Sep 8, 2026
…s now canonical #481 landed on master as c2d35c0. This branch carried #481's original branch commits, so every region reconciled by hand last time came back as a conflict between #481-as-merged and #481-as-this-branch-had-it. Master wins those; only the reachability verdict, the test and the cleared-suspicion comment are layered on top. FIVE HUNKS, AND THE RULE APPLIED TO EACH - ShardManager.onFailure javadoc, the citation to the note #468 retired - MASTER. Both sides had repaired the same dangling reference the same way; master carries the wording added when #481 merged master. - The identical citation in docs/solutions/runtime-errors/retry-queue-write-lock- on-the-rebalance-path.md (add/add - the file is new on both sides) - MASTER, same reason. That file is now byte-identical to master. - ShardManager.onFailure javadoc, the paragraph after "What else the purge now backstops" - OURS. Master has #481's paragraph and nothing else there; this side is the verdict layered under it, which is the content this branch adds. - The inflight note's closing section - OURS. Master ends at "Production reachability is still not established"; this side answers it. - The inflight note's #431 paragraph - OURS, and this is the ONE hunk where master's text is not kept, so the reason is worth stating. Master says "The 2026-09-07 stamp ABOVE says ...". That stamp is not above it here: this branch replaced it with the 2026-09-08 PROPOSED marker, because bin/check-inflight-tags.sh refuses a note carrying two. Keeping master's sentence would leave a pointer to something the merged file does not contain, so this side's wording stands - it makes #481's point, and records what the superseded stamp said so the supersession loses nothing. THE TWO AUTO-MERGES THAT WERE WRONG LAST TIME, RE-CHECKED Both correct this round, because the corrections were already committed rather than being made against #481 for the first time: - ProcessingShard's cleared-suspicion block still reads "the sweep that takes it OUT OF ITS SHARD" and not "from BOTH structures" - the wording #481 falsifies. Its diff against master is exactly this branch's own block and nothing else. - The note carries exactly one inflight-vetted marker. NOT A LOSS, THOUGH IT LOOKED LIKE ONE A sweep for files differing from master that this branch never edited flagged #482's five (BatchTestMethods, CoreBatchTest, KafkaTestUtils and two docs). Cause: #482 landed on master DURING this resolution, so MERGE_HEAD is c2d35c0 and #482 is simply not in this merge. Nothing was dropped; it arrives in the next one. Recorded because "file differs from master and I never touched it" is normally the signature of a silent merge loss, and here it was not. EVIDENCE ShardDisplacementOrphanReachabilityTest 3/3, RetryQueueRebalancePathTest 6/6, RetryQueueRequeueWindowTest 9/9, RetryQueueTest 7/7, RetryQueueIteratorConfinementTest 5/5, ProcessingShardStaleReplacement909Test 4/4, ShardManagerStaleContainerTest 3/3 - 37 tests, 0 failures. bin/check-all.sh 16 passed, 0 failed. Co-authored-by: Claude Opus <noreply@anthropic.com>
6 tasks done
astubbs
added a commit
that referenced
this pull request
Sep 8, 2026
, #481, #482, #485 Two conflicts, and one silent auto-merge artefact that only the compiler found. PartitionState. #469 replaced the pair of booleans with a version counter, deleting stateChangedSinceCommitStart; #410's ledger snapshots at the same point. Joined by taking master's counter whole - the single completionCount load, tested and stashed as completionCountBeingCommitted - and moving the ledger's snapshotForCommit() inside that same branch, still ahead of createOffsetAndMetadata(), so the ledger is still snapshotted before the offer is made and trimmed only on the answer to that offer. The artefact: restoreCompletedButUncommittedWork() still called setDirty(), which #469 deleted, and git merged both sides clean because neither touched the other's lines. It is recordCompletion() now - the replay has moved the partition's state on, so it stamps a new version and isDirty() derives the rest. PartitionStateAbortedTransactionReplayTest 7/7 and PartitionStateCommitWindowSeamTest 3/3 cover the join. The flake ledger takes both sides' rows and master's retirement of the simpleBatchTest section, which #482 fixed - expectation-versus-input, not the library. config/infer-known-findings.txt takes master's version whole. The "1 NULLPTR_DEREFERENCE AbstractParallelEoSStreamProcessor.processWorkCompleteMailBox" line this branch carried was a merge artefact: #346 added it, #466 retired it in the same change that reshaped the mailbox loop, and an earlier merge here re-took #410's older side. The static lane was red on it. Re-ran infer locally on the merged tree - 9 findings, all known, none new. Verified: ProducerManagerTest 22/22, ProducerRecoveryTest 15/15, the PartitionState family, the RetryQueue family, WorkManagerTest, ArchitectureTest, TestConventionsArchTest, TransactionalClaimCoverageTest. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
astubbs
added a commit
that referenced
this pull request
Sep 8, 2026
… and the guard is outside the class (#483) Settles the one open question on docs/inflight/bug-shard-displacement-orphans-the-retry-queue-entry.md: whether production can reach ProcessingShard.addWorkContainer's displacement branch with a queue-resident container, which decides whether the pairing gap there is worth a design change. It cannot. No behaviour changes - the diff is a test, a solutions write-up, and comments. The branch itself IS reachable: confluentinc#909's late drain puts a stale resident in a shard and the next poll's fresh record displaces it. What cannot happen is that resident also holding a retry-queue entry. THE ARGUMENT, EACH LEG TRACED IN SOURCE An orphan needs four things true at once when a container B arrives at (topic, partition, offset): a resident A; A observed STALE, or the arrival is dropped; A holding a RetryQueue entry; and B existing at all. 1. The displaced container is always the one inspected. workMap has two insertion sites - addWorkContainer, which is control-thread-only, and the test-only plantResident. Only removals can interleave between the get and the put, and a removal makes the put return null. A by-key REMOVAL can hit a different occupant (#468's class); a by-key insertion cannot, with one inserting thread. 2. A queue entry implies non-stale-and-resident when it was made. RetryQueue.add has exactly one production caller, ShardManager.onFailure, behind onFailureResult's live staleness check and, since #437, a post-add residency confirmation. couldBeTakenAsWork refuses a stale container outright, so one that is stale on arrival is never selected, never fails, never queues. 3. Staleness is monotone: both epochs are final longs, the fence is set and never cleared, a replaced state carries a higher epoch or is the removed singleton. 4. So A must cross the staleness boundary in between, and only three transitions can do it - the removed-state swap, the assignment putAll, and fenceForRevocation. The first two each unseat A on the same thread inside the rebalance callback; the fence sweeps nothing but cannot be followed by a second container at the same offset, because a duplicate offset needs a re-assignment and that is the swept path. Bumping the epoch map alone is NOT one of those transitions, and this is the leg most easily got wrong by reading: PartitionState's epoch is final and staleness is only asked through the state object, so the window opens at the state REPLACEMENT - which is the same statement as the sweep that closes it. WHAT THIS DOES NOT CLAIM, AND IT IS THE POINT OF THE WRITE-UP The last leg is a property of the KAFKA CONSUMER's fetch position, not of this engine: within one assignment generation the position never goes backwards, so a duplicate offset needs a re-assignment. Nothing in main calls seek, and shards are partition-scoped in every ordering mode. So the pairing gap in that branch is one arrival away from being real rather than absent. An in-generation replay of an already-registered offset - a seek, an offset-reset or truncation replay - makes it reachable immediately, and nothing goes red for that. #481's purge would then collect it within one control-loop tick, which is the second, independent answer: the harm is bounded whatever happens, and this work says the case does not arise. Neither result implies the other. ORDERING-MODE INDEPENDENCE, RAISED IN REVIEW AND CLOSED UNDER TEST Legs one to six never consult the shard key. The mode is visible only in leg seven's second half, and is safe in every mode because a shard is partition-scoped throughout - PARTITION and UNORDERED make it a topic-partition, KEY makes it topic-plus-partition-plus-key. The disproof arm is parameterised over every ProcessingOrder, and a dedicated arm covers the KEY half: two records sharing a key and an offset on different partitions must be held in two shards, not collide in one. Under KEY an emptied shard is garbage-collected, which makes the conclusion stronger, not weaker - a collected shard holds nothing to displace. HOW IT WAS ESTABLISHED ShardDisplacementOrphanReachabilityTest, four arms over six cases, control-armed because a single green arm here is green whether the sweep works or the branch is simply never reached: the production rebalance sequence; confluentinc#909's late drain, which fires the branch with an unqueued container; a positive control that reaches stale-and-resident-and-queued through the production API via the fence and injects the one arrival production cannot supply, where the orphan appears immediately, asserted by reference; and the KEY-ordering collision arm. Ablation, predicted before running and re-run after each of the three merges this branch took, because #481 made the sweeps shard-only and #468 made each removal a compare-and-remove - an ablation against a shape that no longer exists is evidence about nothing. Same result every time: either sweep alone green, both removed red at the load-bearing assertion. The KEY arm was separately verified by sabotage - making the key-ordered shard key topic-scoped turns exactly that arm red, and ShardKeyTest stays green under the same sabotage, which is why the arm is not redundant with it. Two predictions were refuted and are kept visible rather than quietly fixed: that the epoch bump opens the window, and that the replacement would be handed out on the controller pass after the fence - it is not, because the fence makes the whole partition stale. DEFECT-CLASS SWEEP - BY-KEY REMOVALS THAT CAN HIT A DIFFERENT OCCUPANT Reported against the post-astubbs#468 tree rather than fixed here: - ProcessingShard.removeStaleWorkContainersFromShard and getWorkIfAvailable's last-resort sweep - both now conditional on the inspected container. Fixed by #468. - ProcessingShard.onSuccess - still removes by key. #468 reasoned about it explicitly and left it, on the grounds that the container it removes cannot have been replaced. Checked, and left to that argument. - ShardManager.removeWorkFromShardFor - the revoke path's removeWorkAtOffset on the poll thread, and the one remaining unconditional by-key shard removal. Reported, not fixed: it belongs with #468's mechanism, not with a reachability proof. - RetryQueue.remove / removeAll - by coordinates and never by identity, at every call site. Deliberate, but it means a removal can clear a replacement's entry; ShardManager.onSuccess removes by key before it looks up any shard, so a late success for a departed container clears a fresh one's parked-for-retry entry. The cost is an early retry, not a stall or a loss. Noted, not raised as a defect. A cleared suspicion #468 recorded independently at the last-resort sweep rests on the SAME single-writer fact as leg one, reached from the other direction. Both clearances therefore reopen on one ungated condition - anything that puts into a shard off the controller thread - and that is now said at every site that depends on it. WHAT IS LEFT FOR THE OWNER The inflight note stays OPEN with a PROPOSED closed - unreachable AND bounded vetted marker rather than being closed here: its impact is misdirection, and docs/inflight/AGENTS.md reserves that state change for the owner. Co-authored-by: Claude Opus (1M context) <noreply@anthropic.com>
astubbs
added a commit
that referenced
this pull request
Sep 9, 2026
…ever holds the offset (#492) WHAT CHANGED FOR A USER: a rebalance can no longer remove a record that a later assignment had already re-registered. `ShardManager.removeWorkFromShardFor` - the revoke and lost path - removed by KEY with `removeWorkAtOffset(consumerRecord.offset())`, so whatever occupied the offset when the removal landed is what left the shard. It now removes only the container the revoked record was registered as, or a stale one, and never a live container from a later registration. THE DEFECT CLASS, AND WHERE THIS SITE SITS IN IT. #468 closed the same class at the poller's stale sweep: a removal keyed on the offset cannot say WHICH of two containers at that offset it meant, and the fix was to make `WorkContainer` equality identity so that `Map.remove(key, value)` is a true compare-and-remove. #483's defect-class sweep reported this site alongside it - "the one remaining unconditional by-key shard removal" - and deliberately left it, on the grounds that it belonged with #468's mechanism rather than with a reachability proof. This is that. No fork or upstream issue tracks it; it was reported in a commit body and on a `docs/inflight/` note, so this PR closes nothing. THE CALLER HOLDS NO CONTAINER, WHICH IS WHY THE FIX IS SHAPED DIFFERENTLY. The stale sweep inspects a container and then wants that one gone. This sweep is handed the `ConsumerRecord`s ONE generation was still carrying as incomplete (`PartitionState.onPartitionsRemoved` -> `removeAnyShardEntriesReferencedFrom`), so what it can name is the REGISTRATION, not an inspected container. That works because `PartitionState.maybeRegisterNewPollBatchAsWork` puts the same record instance into `incompleteOffsets` and into the `WorkContainer` it builds, in one loop body - so "the container this generation registered for this record" is `occupant.getCr() == revokedRecord`, and a later generation's re-delivery of the same offset is a different object because it came from a different fetch. That identity has one named exception, stated at the site: `addNewIncompleteRecord` puts unconditionally while `addWorkContainer` DROPS an arrival whose resident is not stale, so an in-generation replay of an already-registered offset leaves the partition naming record B while the shard holds the container over record A. The staleness condition below covers it. TWO CONDITIONS, AND THE MIDDLE OPTION WAS WRONG. Registration identity alone declines for a container from another registration that is nonetheless STALE, and declining there would leave a stale container holding an offset of a revoked partition - the state this sweep exists to prevent. Staleness alone declines for a live container built from the record the sweep was handed, which is what a revoke sweep driven before the partition's state has been swapped sees, and is what `ShardManagerLincheckTest.revokeSweep` models on every invocation: that harness's removal operation would have become a silent no-op while staying green. Verified rather than assumed - `incrementPartitionAssignmentEpoch` starts from `KAFKA_OFFSET_ABSENCE = -1`, so that harness's `EPOCH = 0L` containers are live against the epoch its constructor assigns. So the guard declines for exactly one thing - a LIVE container from another registration - which makes the change a no-op everywhere except in the defect case, and each condition has its own ablation arm. THE HARM, IF IT IS REACHED, IS THE LOST RECORD - #468's, not a misdirection. The live container is gone from the shard while its own `PartitionState` carries the offset as incomplete, so nothing selects it and nothing completes it, and the commit high-water mark cannot pass it until the partition is re-polled. The v6 scope note this came from said the cost was "misdirection bounded to one control-loop tick by #481's purge"; that is the DISPLACEMENT orphan's cost, not this one's. This removal is a shard removal, and the purge collects retry-queue entries. REACHABILITY: NOT TODAY, ON TWO INDEPENDENT ARGUMENTS, AND THE STRONGER ONE IS STRUCTURAL. - Thread ordering, which has to name BOTH production routes. On the ordinary route the poll thread runs both rebalance callbacks, so a generation's revoke sweep completes before the assignment that could register a replacement begins. On the CLOSE route the sweep runs on the CONTROL thread (`maybeCloseConsumer` -> `onLeavePrepare` -> `onPartitionsRevoked`), which `PartitionStateManager.resetOffsetMapAndRemoveWork`'s javadoc calls live in every commit mode - and that is also `addWorkContainer`'s thread, so the two still cannot interleave. "Both callbacks are on the poll thread" is the half-statement, and it was written that way here first. - `resetOffsetMapAndRemoveWork` installs `RemovedPartitionState` BEFORE it calls in, so `checkIfWorkIsStale` is true for EVERY occupant and the staleness condition carries all of them: on the shipped path this method is behaviourally identical to the by-key removal it replaces, and the decline branch is defence for a caller that sweeps before the swap. That says nothing about threads. Nothing checks either argument. The conditional form costs one reference comparison and rests on neither, which is exactly why #468 wrote `getWorkIfAvailable`'s last-resort sweep conditionally against a race it had shown was unreachable there. THE NPE GUARDS ON THIS PATH ARE UNTOUCHED, AND THE CLEARED SUSPICION NAMES THE RIGHT OBJECT. The single-read `getShard` idiom #345 put in stays where it was; the new work is inside `ProcessingShard`, below it. The staleness question resolves the partition state from the OCCUPANT, not from `revokedRecord` - and in the only branch that asks, the occupant is by definition a different registration - so the discriminator is two facts: the state is installed before the call (and a partition with no state `continue`s without sweeping at all), and every `ShardKey` variant is partition-scoped, so an occupant of this shard belongs to the revoked partition. WHAT WOULD REOPEN IT: a caller from a path that has not installed a state, or a topic-scoped shard key - the latter being the reopener `addWorkContainer`'s own cleared suspicion already lists, which reopens both at once. EVIDENCE. `ShardRevokeSweepReplacementEvictionTest`, three arms, RED-then-GREEN through `PartitionState.onPartitionsRemoved` so the argument list is built by production code - one level below the state swap, which is what makes the decline branch reachable at all. Ablation matrix, predicted before running, one arm red per ablation and no other: - whole guard removed, back to `removeWorkAtOffset(offset)` -> the displacement arm - staleness condition removed -> `aStaleContainerFromAnotherRegistrationIsStillSweptOut` - registration-identity condition removed -> `aRevokeSweepStillRemovesALiveContainerBuiltFromTheRecordItWasHanded` - decline retires the occupant instead of leaving it -> the displacement arm's accounting assertion That fourth ablation is why the accounting assertion exists: making the decline call `population.onRetired()` + `excludeFromSelection(occupant)` left every other assertion GREEN while creating a permanent `RecordPopulation` deficit, which is this class's recurring defect and has no clamp and nothing to reconcile it. TWO INSTRUMENTATION FAILURES, RECORDED RATHER THAN TIDIED AWAY. The registration-identity ablation reported NO failure on its first run: dropping the condition left an unused local, the module never recompiled, and the grep for failures matched nothing - a silent false negative in the experiment meant to prove the condition earns its place. Re-run with the build outcome asserted, it is red as predicted. Separately, `@SuppressWarnings("ReferenceEquality")` was written on the identity comparison and then measured: deleted, rebuilt under `-Pci`, `BUILD SUCCESS`. Neither `WorkContainer` nor Kafka's `ConsumerRecord` overrides `equals`, so Error Prone never fired - it suppressed nothing and disagreed with `isResident`'s bare `==` two methods away. Removed, with the reason written where it was. SAME-DEFECT-CLASS SWEEP over `ShardManager`, `ProcessingShard` and `PartitionState`. Fixed: the one above; no unconditional by-key removal of a container remains in main. Dismissed with discriminators: `ProcessingShard.onSuccess` (re-verified rather than inherited - `WorkManager.handleFutureResult` is the third staleness checkpoint and drops a stale result before `onSuccessResult`, so a returning stale container cannot remove the fresh one that replaced it); `addWorkContainer`'s by-key INSERTION (#483's leg 1, plus #336's admit-first, which reads the outcome from the map rather than the earlier read); `removeStaleWorkContainersFromShard` and `getWorkIfAvailable`'s last-resort sweep (already conditional, #468); `ShardManager.getShard` / `compute` / `computeIfPresent` (single-read Optional idiom, #345, and atomic per-key, #336 - not check-then-act); `RetryQueue.remove` / `removeAll` at every call site including `onSuccess`, `onFailure` and `purgeDepartedRetryEntries` (same shape, but it is the queue's keying model rather than a removal that forgot to name its target, so a fix is a change to that model - reported and left by #483, and left here); `PartitionState.incompleteOffsets.remove` and the truncation prune (offsets over records, control-thread-only - "this offset is complete" does not depend on which record object represents it). `removeWorkAtOffset` now has no production caller. Kept as the modelling primitive four test classes use for "a container departs its shard", with its javadoc rewritten: it used to open by naming the `onPartitionsRemoved` caller this change moved away, and now says what it is and which of the two conditional removals a main-code site should reach for instead. RECORDS. `docs/solutions/logic-errors/a-by-key-removal-cannot-say-which-container-it-meant-2026-09-07.md` - the class's owner - gains a dated section for the second site, because its own sweep's only hit was a false positive and the real one was found a day later. `docs/inflight/bug-shard-displacement-orphans-the-retry-queue-entry.md` gets a pointer, not a narrative: its subject, state and `PROPOSED closed` marker are untouched, and what is recorded is only that a reader arriving at #483's four-item list can see which one moved. Reproduce: `bin/ci-unit-test.sh`; `bin/lincheck-test.sh`; `bin/check-all.sh`; `./mvnw -pl parallel-consumer-core -am test -Dtest='ShardRevokeSweepReplacementEvictionTest,ShardStaleSweepReplacementEvictionTest,ShardManagerRevokeSweepNpeTest,ShardDisplacementOrphanReachabilityTest'`; `bin/check-pr-analysis-surfaces.sh 492`. Co-authored-by: Claude Opus (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
No fork issue exists for this - the defect was found by a defect-class sweep and tracked only as an in-flight note, which this PR retires.
This supersedes #431, which fixes the same defect and should not be merged. That PR is correct and proven; it is more machinery than the defect needs. It is not closed here - that is the owner's call. See "Why this rather than #431" below.
Description
The defect
RetryQueue.removetakes an unbounded, FAIRwriteLock().lock(), and every Kafka rebalance callback reached it on the broker-poll thread:ShardManager.removeWorkFromShardFor, on the revoke and lost paths;.map(retryQueue::remove)) throughShardManager.removeStaleContainers, on the assigned path as well as those two.The controller thread holds the matching read lock for a whole scan -
RetryQueue.iterator()hands it out and it is released only at close, whichShardManager.getLowestRetryTimedoes around an entire iteration. The lock is fair, so an arriving writer queues behind that scan rather than interleaving with it. That wait sits insideconsumer.poll()with the whole group waiting on it, spent out ofmax.poll.interval.ms. No cycle was found, so this is an unbounded wait, not a deadlock.The fix
The poll thread never touches the retry queue. The controller thread collects what it leaves.
ConcurrentHashMapofConcurrentSkipListMaps, with no lock to wait for and so nothing to decline.ShardManager.purgeDepartedRetryEntries()collects, on the controller thread where waiting is allowed, every retry-queue entry whose containerProcessingShard.isResidentsays has left. It scans under the read lock, closes the iterator, then removes the collected entries under the write lock in oneremoveAll.The invariant, stated once at the purge: a retry-queue entry with no resident container is garbage the controller collects; it may exist for at most one control-loop tick; the callbacks are free to create it.
Where it runs: at the top of
ShardManager.getWorkIfAvailable, the one point the control loop reaches unconditionally every pass throughretrieveAndDistributeNewWorkin bothRUNNINGandDRAINING, and beforedrain()readsgetNumberOfWorkQueuedInShardsAwaitingSelectionin that same pass. The drain path is the consequence that matters: #437 established that a queue-only orphan survives that figure's flooring once the pipeline is drained, so it reads true forever and a draining close hangs to its timeout.The bound is one control-loop tick in every ordering mode. The purge scans the retry queue rather than the shards, so the ordered modes' shard-scan break - which stops the scan at the first container handed out - cannot delay it. This is the one place the shipped design is strictly better than #431's rather than merely smaller: that one's bound was "until the takeable head in front of it leaves the shard".
Why the scan and the removal can be two steps. Departure is monotonic (
isResidentis reference identity, and nothing re-inserts the same instance), and nothing else writes the queue's keys:RetryQueue.addis@ControllerThreadOnly, its only production caller isShardManager.onFailure, and the purge runs on that same thread. The second half matters becauseRetryQueueremoves by topic/partition/offset and so cannot say which container it meant - the defect class of #468. What would reopen it is a second writer of the retry queue.Why this rather than #431
That PR keeps the poll thread on the queue and teaches it to decline:
tryRemove, abandon the paired shard removal on refusal, and (after #437) a second ask after the shard removal with a put-back on refusal. Over the same three main-code files it is+347 -72against+185 -67here, and most of the latter is javadoc stating the invariant. What those lines buy is the part that matters: a second non-blocking entry point onRetryQueuekeyed differently from the blocking one, a queue reference threaded into a secondProcessingShardmethod, two asks per removal, and a retirement bound that varies by ordering mode.Also rejected, by name: reusing #466's control-thread hand-off. That mechanism has the poll thread wait on a
CompletableFuturefor the controller to serve a posted request - right for a revocation commit, which is not correct until the commit has happened; wrong here, because the purge needs no hand-off at all (the controller discovers departed entries by scanning) and reusing that shape would reintroduce the very wait being removed. A dirty flag set by the callbacks was also rejected: a second piece of cross-thread state on a class whose whole difficulty is cross-thread state, for a scan the control loop already performs on the same structure.Counters during the one-tick window, re-measured on this tree
Between a callback's shard removal and the purge,
getWorkableRecords(inShards - parkedForRetry) reads one LOW - negative in the single-record case. Low is the safe direction:isSufficientlyLoaded()isworkable > threshold, so it fetches sooner; high would pause the poller with nothing to process, which is the confluentinc#857-family stall shape.WorkManagerTest.theLoadGateReadsLowNotHighForTheTickAfterARevocationpins the direction and that one controller pass settles it. This re-confirms what #437's squash body measured about the same figure.What #437's residency confirmation became
ShardManager.onFailureadds to the retry queue and then confirms shard residency, undoing the add if the container has left. Against a shard-only sweep it still closes two of the three interleavings; the third - the sweep starting after the residency read - is now collected by the purge a tick later. So it is belt-and-braces, kept this round, with its javadoc saying so and naming it as a live question rather than a settled one. Removing it is a separate decision with its own evidence to gather.Its test
RetryQueueRequeueWindowTest.theProductionSweepsQueueRemovalIsGatedOnItsShardRemovalasserted a property of a sweep that no longer exists. It is re-stated rather than deleted, astheProductionSweepLeavesTheRetryQueueToTheController: the sweep touches no queue at all, and the controller pass afterwards collects exactly the departed entry and leaves the resident one - so it still goes red both if a callback starts touching the queue again and if the purge stops discriminating.theSameRebalanceOneSeamLaterTakesTheQueueEntryWithTheShardEntryis re-stated the same way.aQueueFirstSweepDefeatsTheOneShotConfirmationkeeps its name and its assertion, because two write-ups cite it; its javadoc now says it models a superseded design rather than an incoming one.Evidence
Red first, then green.
RetryQueueRebalancePathTestholds the read lock exactly the way the controller thread does - through a liveRetryQueue.iterator()- and drives the production callback on a thread namedbroker-poll. The harness is adapted from #431, including its load-bearing join placement. Againstorigin/master, five of six arms fail:anEntryWhoseContainerIsStillResidentIsNotCollectedis the control and is green throughout. With the fix, all six pass.Ablations. Deleting the purge call turns six arms red across three classes. Inverting the residency predicate turns seven red - the same six plus the control, which is what the control is for.
The ArchUnit rule was proven on merit, not assumed. With
KNOWN_BLOCKING_VIOLATIONSemptied and the callbacks still reaching the queue,rebalanceCallbacksMustNotBlockreports 24 violations (18 exemption keys; the@ControllerThreadOnlyhalf fires once per route, so the revoke and lost roots report through bothremoveWorkFromShardForandremoveStaleContainers). With the callbacks off the queue it is green with the set still empty. All eighteen entries are deleted. The set, itsroot => targetkeying and the argument for keeping it remain: an exemption is how a newly-found reach is recorded as a defect on the books rather than deleted from the report.RetryQueueLincheckTestneeds no new operation, and that is a claim rather than an omission: this PR adds no new production write path toRetryQueue. The purge composesiterator()and the existingremoveAll, both already modelled.ShardManagerLincheckTestand the whole Lincheck lane (bin/lincheck-test.sh) were run and are green.Gates run locally, in the foreground:
bin/check-all.sh(16 passed, 0 failed),bin/lincheck-test.sh(green),bin/ci-unit-test.sh(green across every module). No Docker-backed tests were run - the box is short of disk - so the integration lanes are CI's.Defect-class sweep - other instances of "a rebalance callback waits"
ArchitectureTest.rebalanceCallbacksMustNotBlockis the mechanical sweep for this class and is now green on merit with no exemptions, which is a stronger answer than a grep. What it cannot see is stated in its own javadoc and inControllerThreadOnly's: asynchronizedBLOCK is aMONITORENTER, invisible at any depth (PCMetrics.removeMeter's monitor andclearCommitCommand()'s are the two known ones, both left alone deliberately); a stored-rather-invoked method reference has the same shape as an invoked one; and dynamic dispatch through a user-suppliedConsumerRebalanceListeneris out of reach. None of those is changed here. Infer's@Locklessis the only tool that can enumerate the monitors, and its verdicts are already recorded in the coreAGENTS.md.Simplify pass
ce-simplify-code, three reviewers over the diff. Two findings applied, one recorded rather than applied.Applied - the new test hand-rolled a harness
BlockedThreadAsserteralready is. It built its own thread, latch, bounded await and join. That class gains a third idiom,returnsWithin(function, threadName, budget)plusjoinQuietly- the negative of the two it had, because a caller proving a function does NOT block has to keep holding the thing it would wait for and read state while the contention is on, so it cannot give up its own thread to an assertion helper. Beyond deduplication this buysrunCapturingThrowable: a callback that throws on entry returns promptly and was otherwise indistinguishable from one that did its job, which for a test asserting promptness is a false pass. Re-proved after the refactor rather than assumed - restoring the queue removal to both callback paths turns three arms red again, the two lock ones by timing out;BlockedThreadAsserterTestandProducerManagerTestare green.Applied -
ShardManager.onFailure's residency branch described a race that no longer exists. Its comment and log read afalsefromretryQueue.remove(wc)as "the revoke sweep's own paired removal got there first". That cannot happen now, sotrueis the only outcome production can produce andfalsebecame a signal: it would mean a second writer of the retry queue exists, which is the one thing that invalidates the purge's scan-then-remove. Said that way at the site.Recorded, not applied - the purge is the first unconditional full scan of the retry queue on the control loop. The quality and efficiency reviewers found this independently. Every other reader there early-stops on the
retryDueAtsort order; shard residency has no relationship to that order, so this one cannot, and it finds nothing on most ticks. The review also sharpened the fix: a monotonicLongAdderon departure, in the shapeRecordPopulationalready uses at those very call sites - not the boolean dirty flag the design rejected - which keeps the one-tick bound trivially.Not applied here for two reasons, both stated in the write-up's rejected alternatives and the branch note: it puts poll-thread-written state back on the rebalance path in the change whose point is taking the poll thread off this structure, and the cost is unmeasured.
DispatchScanMeteris this class's own precedent for metering exactly this, and the honest order is meter, then decide. Queued as a decision rather than lost as a review comment.Also confirmed and deliberately not acted on: the per-tick
ArrayListallocation is free (the no-arg constructor defers the backing array), and the fair read lock held across the scan cannot hurt the poll thread any more, because the poll thread no longer touches that lock at all.Collisions
Textual only; none conflicts in design. #468, #431, #410, #392 and #362 all touch
ShardManagerorProcessingShardaround the sweep; #408 touchesArchitectureTest. Whichever lands second resolves the text.Records
docs/solutions/runtime-errors/retry-queue-write-lock-on-the-rebalance-path.mdis new and records both designs, adapted from the write-up fix(core): the rebalance callbacks decline the retry queue's write lock instead of waiting for it #431 wrote for itself so its diagnosis is preserved rather than re-derived.docs/inflight/bug-retry-queue-write-lock-on-the-rebalance-path.mdanddocs/inflight/pr-431-must-pair-its-queue-removal-with-the-shard-removal.mdare retired into it.static-archunit-main-code-rules.md,test-retry-queue-behaviour-untested.md,bug-shard-displacement-orphans-the-retry-queue-entry.md,core-retry-queue-needs-a-runtime-controller-ownership-guard.md,test-largenumberofinstances-residual-failures-measured-not-explained.md,static-infer-threadsafe-is-blocked-by-third-party-interfaces.md, anddocs/solutions/runtime-errors/retry-queue-orphan-window-between-the-requeue-check-and-the-add.md.docs/inflight/branch-retry-queue-controller-purge.mdis the working note.Checklist
docs/features/- N/A - a concurrency defect fix with no user-facing feature or option; the observable change is that a rebalance is no longer delayed by the control loop's retry-queue scanRetryQueueRebalancePathTest(new, red-first), plus re-stated arms inRetryQueueRequeueWindowTestandWorkManagerTest, and the emptiedKNOWN_BLOCKING_VIOLATIONSdocs/inflight/working note (pr-/branch-) started at the PR's first commit -docs/inflight/branch-retry-queue-controller-purge.md; it will be renamed topr-if the number is worth carryingce-simplifyandce-code-reviewlocally -ce-simplify-coderan on the diff vsorigin/masterwith all three reviewers (reuse, quality, efficiency); its findings and dispositions are in the "Simplify pass" section above.ce-code-reviewwas NOT run locally:@claude review thisis asked for on the PR instead, which is the route that can open inline review threads and so the one that mechanically blocks the merge🤖 Generated with Claude Code
https://claude.ai/code/session_018rEzrWYFr6oEzy6porczd3