Repository navigation
refactor(core): confine the consumer to its owning thread, and make ownership a lifecycle - #393
Merged
Merged
Conversation
…wnership a lifecycle Extracted from #29 and cut fresh from master, where it stands on its own. Its decomposition plan called for exactly this and said why: it is the largest piece of that branch by main-code volume, and its shape is a design decision about consumer ownership rather than a patch, so it earns separate review. WHAT IT IS KafkaConsumer is not thread safe, and this codebase runs two long-lived threads near it. ThreadConfinedConsumer wraps the consumer and refuses any call from a thread that does not own it; ConsumerOwnership makes that a lifecycle rather than a one-shot flag - the poll thread claims at start of life, releases when the poll loop exits, and a closing thread may claim only what is unclaimed or already its own. Never stealing is the point: a guard that let close bypass the check would legalise closing mid-poll. WHY IT IS SEPARATE FROM THE DEADLOCK FIX The measured claim on #29 is about roughly twenty lines of lock change - a revoke path that declines with tryLock instead of blocking - verified by a four-cell control on both assignors. That control never measured this code. Reviewing the two together invites the verification of one to be read as evidence for the other, and they are independent: the deadlock fix references this wrapper only in comments, and compiles without it. IT CARRIES A REGRESSION IT CAUSED, AND THE TEST THAT PROVES IT Restructuring ConsumerManager for thread confinement replaced a per-poll updateCache() at poll ENTRY with one-time priming in init(). That left the pause cache written only when a poll ended - so during every paused long poll the control thread's back-pressure wakeup read the previous poll's "not paused" value, never fired, and each pause cost a full long-poll timeout with the pipeline drained. Measured on the 857 branch as a four- to tenfold transactional throughput collapse that read as a flaky lane for weeks. The entry refresh is restored here, deliberately BEFORE pollingBroker.set(true) rather than after it as on master, so wakeup() cannot reach the consumer while these calls run - preserving the thread-confinement property this change exists for. ConsumerManagerPauseCacheTest holds that contract and is RED-PROVEN on this branch, not assumed: removing the entry refresh fails it with "expected: 1", restoring it passes. MockConsumer's schedulePollTask observes the cache from inside the poll it describes, so the test is deterministic - no timeouts, no concurrency. VERIFIED All sixteen tests across the four suites pass on master with only these files added; main and test compile clean; bin/check-all.sh 15/15. WHAT DELIBERATELY DID NOT COME ArchitectureTest stays on #29. It holds this work's consumer-holder allowlist AND that branch's rebalance-blocking rule, and splitting a shared file would drag unrelated work in either direction. Nor did the docs. One test cited a solution write-up that lives on that branch, and following the citation pulled a plan, then two inflight notes, then bin scripts - an interconnected research trail, not this change's rationale. The citation is now by topic rather than by path, because a path that does not resolve on the branch it is read from is worse than no citation.
Merged
4 of 6 tasks
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 |
🧪🔒 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 |
|
astubbs
added a commit
that referenced
this pull request
Aug 31, 2026
…and migrate the #119 answer somewhere it survives Two jobs, and the second is the one that would have cost something. AUDITED: the "Still open" list was written 2026-08-19 and three of its items have since been done * The review deferral's PRECONDITION IS MET. It was deferred until #204, #31, #57 and #267 landed - all four have, master has been merged in, and the deadlock fix was re-verified afterwards on a four-cell control across both assignors. Review is now the right next step rather than a wasted cycle. Two NEW parents have appeared since that order was written - #381 and #393 - and the note says explicitly that neither is a reason to defer again, because neither changes this tree. * The roadmap edit is done, and the constraint the item stated is respected: it was not to be edited early in a way that asserts something not yet true, and the new wording says the fix is measured and unmerged, which is what is true now. * The merge strategy is recommended and the squash message written, in the scratchpad rather than the PR body. MIGRATED: the #119 answer now lives in the family ledger That answer existed only in this PR's own note, which is DELETED when the PR lands - so the established answer to "does #119 close with this?" would have gone with it, and the next person would have re-derived it from the same sightings. It is now in bug-857-family.md, which is on master and is the durable owner of this family's accounting. The answer, briefly: no, because #119 is a symptom bucket rather than a defect. The AB-BA cycle the fix breaks cannot close in PERIODIC_TRANSACTIONAL_PRODUCER at all - its second edge lives in a committer built only for the consumer-commit modes - which leaves the unbounded transactional revoke wait, #44, the only issue upstream ever labelled a verified bug, untouched. Closing #119 here would retire the remainder silently, which is the specific damage the merge checklist names. One mechanical note for whoever edits that ledger next: it carries a file-wide post-merge checked-begin/checked-end block, so a new section inside it must NOT bring its own pair. A nested pair breaks the outer one and the gate then flags lines far above the edit, which is a confusing way to find out.
astubbs
added a commit
that referenced
this pull request
Aug 31, 2026
…efect, and the extraction already ruled one suspect out This note predicted that the Performance Tests failure shared a cause with the Integration Tests one - the stale pause cache - on the reasoning that the mechanism is commit-mode-independent. The lane has now run with that fix in place and still fails, so the prediction is REFUTED. Recorded as a refuted prediction rather than replaced by the new number, because a prediction that did not hold is worth as much as one that did and vanishes if only the outcome is written down. The measurement, same test and identical configuration: #381 processed=3000000/3000000 elapsedMs=42024 71387 rec/s PASS #29 processed=2191095/3000000 elapsedMs=60554 36184 rec/s FAILED #381 is test-scope and cut from master, which makes it the closest thing to a master baseline the lane can produce - it does not run on master. Getting that baseline by riding the reporter through on a PR the failing one already depends on was the point of the extraction, and it arrived in time to stop a silent halving from merging. Two samples on one side against one on the other, all on hosted runners: enough to attribute, not enough to treat the ratio as precise. AND ONE SUSPECT ELIMINATED FOR NOTHING #393's performance lane passes, and its thread-confinement refactor is also in the failing tree. So the consumer-ownership work is not the cause, established without running an experiment. The general form is worth more than the result: a branch cut out of a suspect tree is a control arm for whatever stayed behind. Check whether the decomposition already gave you one before designing a bisect. The cheapest untried ablation is named - the revoke fork's two log lines were raised to INFO, and per-loop INFO logging in a three-million-record run is the right shape to cost this much throughput. The coherence check and the pause-observation fields are on the same list. One term at a time, same lane, control arm. Numbers here are anchored to the runs that produced them and written in the past tense, not as standing properties, so they read as evidence rather than as current state that nothing will update.
…first review First review and simplify pass over this branch - it had never had one. PR #393 carried zero claude[bot] comments, and the green claude-review on #29 predates this code by three weeks, so there was no prior review to inherit. Three reviewers (reuse, quality, efficiency) ran over the branch diff; what follows is what survived scrutiny. DELETED A MIRROR THE BASE COMMIT HAD JUST ARGUED AGAINST assignmentSizeCache, getAssignmentSize() and the consumer.assignment() call inside updateCache() are gone. The field was a fresh copy of a fact Kafka owns, its only readers were one log line and a getter nothing called, and it made updateCache() ask the consumer three times instead of two - twice per poll, since this branch refreshes on entry as well as exit. assignment() takes the consumer's lock and copies the assignment set to a new HashSet before .size() throws it away. This is not a preference. The commit this branch sits on, b2e6c19, deleted pausedForThrottling for being exactly this shape and shipped docs/solutions/architecture-patterns/a-mirror-of-state-another-component-owns-is-a-contract-nobody-wrote.md to name the class - "a mirror is usually a threading constraint wearing a field's clothes" - and its own body records that the poll hot path should "read Kafka once per pass, not three to five times". Adding an unread mirror in the next commit would have reopened the case that document closed. If #29 needs a control-thread assignment size, it should arrive with the reader that justifies it and the contract the document asks for at the field. The per-poll log line goes back to log.debug from log.trace, at the level it has always been. The demotion was not a decision, it was fallout from appending the assignment-size argument to that line. DELETED getConsumerClass(), WHICH COULD NOT HAVE WORKED Its javadoc says it exists "for reflection-based checks (e.g., auto-commit detection)", but the `consumer` field is a ThreadConfinedConsumer now, and getClass() is final on Object - so it returns ThreadConfinedConsumer.class and never KafkaConsumer, MockConsumer, LegacyKafkaConsumer or AsyncKafkaConsumer. Every branch of AbstractParallelEoSStreamProcessor#getAutoCommitEnabled tests exactly those names and then reflects into fields that exist only on the real driver, so wiring this in would have silently failed the auto-commit-disabled check rather than reporting anything. Nothing calls it today, so removing it is behaviour-preserving and the trap goes with it. Whatever needs the driver's type must unwrap the delegate, not ask the wrapper. The other unwired members stay: claimConsumerOwnership, releaseConsumerOwnership and the two subscribe overloads are package-private, correct, and the seam #29 wires in - splitting them out buys a second review of code that will be reviewed again when it gains a caller. TWO GUARD-INTERNAL FIXES, ONE OF THEM IN THE REPO'S NAMED DEFECT FAMILY releaseOwnership's failure log re-read `ownership` instead of reporting the `witness` its own branch had tested. When the CAS is what failed, the message names an owner the method never saw and says the caller does not hold something it held a moment earlier - the two-read divergence parallel-consumer-core/src/main/java/bz/stub/parallelconsumer/AGENTS.md calls out as its own class, in the one place a reader turns to when the guard misfires. claimOwnership's `previousOwner != null && previousOwner != current` had a conjunct that cannot be false: isOwnedBySomeoneOtherThan already established it. The null sentinel and the ternary that built it are gone with it. ONE TEST FIXTURE INSTEAD OF TWO COPIES ThreadConfinedConsumerTest and ConsumerManagerCloseOwnershipTest each built a MockConsumer behind a ThreadConfinedConsumer plus a named foreign thread - the clone both duplication engines flagged on the PR. They now share ThreadConfinedConsumerTestBase. The copies had already drifted in the way that matters: only one unwrapped ExecutionException, so an assertion failing on the foreign thread surfaced in the other as a wrapper naming neither the assertion nor the thread. The shared version unwraps, and the foreign thread's name is a constant the rejection-message assertion reads rather than restates. WHAT WAS LEFT ALONE, AND WHY The entry refresh before pollingBroker.set(true), the never-steal rule, RELEASED as a state distinct from UNCLAIMED, ConsumerOwnership being top-level rather than nested, wakeup() staying undelegated, close() logging rather than branching on a failed claim, and the 19 duplicated signatures in the Lombok excludes interface - all defended in the extraction commit or forced by @DeleGate, all untouched. VERIFIED ConsumerManagerPauseCacheTest re-proven RED against this diff, not assumed: with updateCache() now making two consumer calls instead of three, removing the entry refresh still fails it with "expected: 1", and restoring it passes. The 16 tests across the four suites still pass; the full core unit suite is 546 passed, 0 failed, 8 skipped; main and test compile clean; bin/check-all.sh 15/15. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01UH2q829FDKQcqQseE8Cp7q
astubbs
added a commit
that referenced
this pull request
Aug 31, 2026
… branch calls - record the merge decision before it becomes a surprise The simplify pass on #393 removed assignmentSizeCache, ConsumerManager.getAssignmentSize() and the consumer.assignment() call in updateCache(), having established there were no callers. That is true on that branch and the reasoning is sound - the commit it sits on deleted pausedForThrottling for being the same shape, and the poll hot path is meant to read Kafka once per pass, not three times. It is not true of the merged result. This branch adds a public AbstractParallelEoSStreamProcessor.getAssignmentSize() over the deleted method, and MultiInstanceRebalanceTest calls it in a failure-time state dump. A "no callers" check run inside one branch cannot see the branch that will merge it, which is the general form worth remembering: after extracting work into a dependency, a deletion justified by zero callers has to be re-asked against every branch that will merge it back. Recorded rather than fixed, because the fix belongs to the merge and the merge is not yet due - #393 has not landed. It fails loudly as a compile error when it happens, so this is a note to save the next session ten minutes, not a defence against silence. The note also says what the resolution is NOT: reverting the deletion. The surviving caller is a diagnostic on a failure path, called once per dump, so the per-poll cost that justified removing the cache does not apply to it - read consumer.assignment().size() at the point of use, or drop the field. Nothing on master calls either method, so no shipped API is involved.
…rule, not vigilance ThreadConfinedConsumer guards by overriding each Consumer method it cares about and naming that method on a Lombok @DeleGate excludes interface, so no competing passthrough is generated. The two lists must agree and nothing checks that they do. Drop a signature from the excludes interface and the generated delegate wins: the call reaches the consumer without the ownership check, with no compile error and no failing test. A new Consumer method arriving in a Kafka client upgrade is the same hole approached from the other side - it is delegated by default, and a default is how a guard acquires a gap nobody chose. Added to the existing candidate list in the ArchUnit note rather than as a new note, because that file already owns this decision and a second list would drift from it - the same reasoning the note itself applies to frozen baselines. Two conditions recorded with it. It is the per-invariant exception the note already allows, not the post-v6 programme, since it pins an invariant that arrives with the work rather than policing structure. And it must wait until the guard is actually wired, because a rule pinning an unenforced seam pins nothing - ownership never leaves UNCLAIMED on this branch, which allows every thread. Suggested by the simplify pass on this wrapper. Not created as a tracked task: whether it is worth the rule is the maintainer's call, and the note is where such calls are weighed.
astubbs
added a commit
that referenced
this pull request
Aug 31, 2026
…t not come back the way it left #393 removed ConsumerManager.getConsumerClass(), which this branch still defines. Recorded next to the getAssignmentSize decision because both surface in the same merge and they want opposite handling. The method could never have worked. consumer is a ThreadConfinedConsumer and Object.getClass() is final, so it returned the wrapper's class - never KafkaConsumer, MockConsumer, LegacyKafkaConsumer or AsyncKafkaConsumer. Its purpose was reflective auto-commit detection, which string-matches precisely those four names and then reflects into fields only a real driver carries. That is what makes it worth a line here rather than a shrug. Having no auto-commit check is a gap someone eventually notices. Having one that always takes the not-a-known-driver path looks like a check, passes, and reports nothing - so restoring the method during conflict resolution would leave the branch better defended on paper and no better defended in fact. If the capability is wanted, it has to unwrap the delegate. The note says so at the point where the conflict will be resolved, because that is the moment someone chooses a side, and the version that compiles is the wrong one.
Owner
Author
|
@claude review this |
|
Claude finished @astubbs's task in 7m 15s —— View job Review complete
SummaryThis is a clean, well-reasoned extraction. The ownership lifecycle ( I ran the four new/changed test classes locally ( I also verified the honesty claim in the PR body: The pause-cache fix logic checks out: Findings1. Dead code: two new
Given the PR's own "no half-finished implementations" discipline and that everything else here is scoped tightly to what's actually reachable, this pair looks like it was added in anticipation of #29 wiring 2. Minor: Not re-litigatedEverything the PR body explicitly scoped out checks out as scoped out: |
The automated review read the two new ConsumerManager.subscribe overloads as dead code and asked for them to be removed or explained. It was right that nothing calls them, and right that nothing at the code said so - the PR body's "guard installed, not guarding" paragraph is where that fact lived, and a PR body is not where the next reader of this file will look. So the reviewer reached exactly the conclusion that paragraph exists to prevent, which is the evidence that a body-only note was not enough. Keeping them rather than deleting: they are the seam #29 wires, where the processor is changed to hold this manager instead of the raw consumer and the ArchUnit rule pinning the invariant lands with it. A prior simplify pass on this branch already kept them for that reason. Deleting and re-adding is churn that also loses the reasoning a second time. The note now sits on the methods, so the next reader gets the answer where the question occurs to them. The review's second finding was a real inaccuracy and is corrected here. The init() javadoc offered a missing group.id as the example of what its catch absorbs, but the processor's constructor runs validateConfiguration() - and so checkGroupIdConfigured() - before it asks the module for the broker poller, and the poller is what first constructs this manager and calls init(). On that path the missing-config error has already been thrown, naming the config, before priming is reached, so the stated example cannot occur. The catch is justified by other priming failures and by the manager being built lazily, which is what the javadoc now says instead. Comments only - no behaviour changes.
… at it Found by bin/check-pr-analysis-surfaces.sh, not by the review: SpotBugs (fb-contrib HES_EXECUTOR_OVERWRITTEN_WITHOUT_SHUTDOWN) fires on the foreignThread assignment in this fixture's @beforeeach, and it fires on a line this branch wrote - the category AGENTS.md says you fix or explain, and which nobody had read here. The static lane is non-gating, so it went green and said nothing, which is the #356 shape that script exists to catch. The named harm is not present: the pool's threads are daemons, so a leaked one could not hold a JVM open, and JUnit runs the teardown between tests so the field is never actually overwritten while live. But overwriting a live ExecutorService field is a leak shape regardless of who currently guarantees it cannot happen, and the guarantee here is JUnit's lifecycle rather than anything this class states. Shutting down what is about to be dropped is simply correct. Chose this over the registry route deliberately. docs/inflight/static-spotbugs-rule-registry.md sanctions suppression by RULE, in a checked-in filter, which would switch HES off for the whole repository and lose every future instance in new code - and it prices that at "the count of off-rules only goes down". Three lines at one site is the cheaper answer. Verified rather than assumed: SpotBugs over the module at the CI settings reported the finding before and does not report it after, with test sources demonstrably in scope (test classes appear among the other 303 findings). Same-class sweep: HES_EXECUTOR_OVERWRITTEN_WITHOUT_SHUTDOWN has exactly one other instance in the reactor, ChaosConductor$ChaosConductorBuilder.pcExecutor, which also carries HES_EXECUTOR_NEVER_SHUTDOWN. It is pre-existing on master in a file this branch does not touch, so it is inherited and stays for the registries rather than being bulk-fixed here.
Owner
Author
|
Both findings addressed, plus one the review did not reach. Pushed as d9e3614 and de74f62. 1. The unwired
|
… key collision
The Unit Tests lane went red on ReactorBatchTest.simpleBatchTest under KEY
ordering with the ledger's exact signature - Expected size: 3 but was: 4. That
made it a sighting to record before the log expires, which is the rule the
ledger exists to serve.
The payload is the reason this is worth more than a tally. The five records
carried keys 34, 62, 34, 34, 77 and arrived as {o0,o1}, {o4}, {o2}, {o3}: a
three-way collision on one key, two singletons, four batches against an
expectation of ceil(5/2). The 2026-08-25 sighting carried keys 29, 36, 36, 36,
71 arriving as {o0,o4}, {o1}, {o2}, {o3} - a different module, a different
collided key, the same shape. The ledger had called the expectation-versus-input
reading a lead nobody had tested; it now has two independent draws behind it,
which promotes it from one of three equal readings to the leading one. The
experiment that settles it is unchanged and still unrun.
Also restated the master-state argument rather than reusing it, because this is
the first sighting on a branch carrying main Java and the previous entries all
leaned on "no main Java" to rule out the PR. The narrower argument: relative to
the head the failure first appeared on, this branch's diff is comments plus one
core test fixture that ReactorBatchTest never loads, and its one poll-path change
reorders an existing updateCache() call rather than adding work.
The whole section is written in post-merge terms and marked accordingly, so it
still reads correctly once the branch it was found on is gone.
astubbs
added a commit
that referenced
this pull request
Sep 1, 2026
…on Linux - it fails, and the probe says a member went silent THE RATE EXISTS NOW, AND SO DOES A FAILURE One failure in ten consecutive runs at the default scale, plus a passing pilot, on an otherwise idle Linux box whose /etc/environment pins ActiveProcessorCount=8. That is the first time this test has failed anywhere but CI - the nineteen green runs recorded in August were all on an M2 Mac Pro, and a machine that never fails cannot separate a coordinator problem from a PC one. IT IS A STALL, NOT THE OVERLOAD THE AUGUST SWEEP MEASURED The detector's verdict was FLAT - "it stopped rather than ran out of time" - at 432,799 of 500,000 records with 12s of no movement. The confounded 2026-08-28 sweep failed the opposite way, with TimeoutException at scales the machine could not finish. That difference is the whole reason this run is evidence and that one was not. THE PROBE MOVES THE STANDING CLAIM, WITHOUT SETTLING IT The ambient probe caught ZOMBIE_MEMBER/REBALANCE_BLOCKED: the group dwelt in PreparingRebalance for 15s because "a member is not answering the rebalance", with the commit path full of RebalanceInProgressException for the duration. This note's title claim - that the residual is the Kafka group protocol under churn - describes a coordinator that cannot converge. What was observed is a coordinator blocked on a silent member, and the members are PC instances. It is not proof. The capacity profile kills instances continuously, and an instance killed mid-JoinGroup is by construction a member that stops answering. But the open question is now much sharper: is the unresponsive member one the harness stopped, or one still running? THREE INSTRUMENTATION GAPS STAND BETWEEN THAT QUESTION AND ITS ANSWER ProgressTracker.withDiagnostic has never been called anywhere - the hook exists, its javadoc names pc::describeProgress, and every stall this test has ever produced ends with "no consumer diagnostic supplied". dumpInstanceState deliberately cannot report the assignment since #393 removed the mirrored accessor. And probes.log and pc-poll-rebalance-lifecycle.log are empty on every passing run, so a failing run has no passing run to be read against - whether a 15s dwell is anomalous or routine under this churn is not answerable from what the runs record. THE HANDOFF'S "NO MEASUREMENT AT ALL" WAS WRONG WHEN WRITTEN Corrected beside the original rather than over it. Nineteen runs were already recorded in the very note the handoff cited as owning the question. What was true is narrower: the experiment script had never been run, having landed the same day. "The instrument has never been run" and "the question has never been measured" are different claims, and conflating them buried a result and sent an earlier session hunting CI history for a number that was one grep away.
astubbs
added a commit
that referenced
this pull request
Sep 1, 2026
…all four were decided Its status line said "cluster 1 measured, clusters 2-4 undecided". That stopped being true on 2026-08-27 and nothing updated it, so a reader arriving today would think three open decisions were waiting for them. WHAT LANDED, WITH THE COMMIT FOR EACH Cluster 2, ThreadConfinedConsumer, became #393 (951ae1e) - the plan said do not land as-is, last and separately, and that is exactly how it went. Cluster 3, the counter adjustment, became #336 (3e668a4). Cluster 4, the pausedForThrottling reset, became #376 (b2e6c19). Cluster 1 is this PR and is still open. THE OUTCOME WAS BETTER THAN THE PLAN IMAGINED, IN THE SAME WAY TWICE Clusters 3 and 4 were framed here as "half a fix" and "probably a regression", which invites patching or reverting. Neither happened. Each turned out to mirror a fact something else already owned, and both were fixed by deleting the mirror: the load gate is now derived by conservation, and the back-pressure pause is read from Kafka instead of tracked alongside it. Recorded because the plan's framing would have led somewhere worse, and that is worth a future planner knowing. WHAT WAS NOT DONE The document is not rewritten to match today's code. The decomposition it argued is the reason those three PRs exist, and anyone arriving from one of them needs the reasoning as it stood on 2026-08-18. Only the status line and a closing section were added, per docs/citations.md's rule about dated records. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01UH2q829FDKQcqQseE8Cp7q
astubbs
added a commit
that referenced
this pull request
Sep 1, 2026
…o its instrument branch THE PLAN IS DELETED, AND THE ARGUMENT FOR KEEPING IT WAS WRONG It was defended here on the grounds that a reader arriving from one of the three PRs it produced would need its reasoning. Checked: none of #393, #336 or #376 references it. Its only durable citation was one line in the architecture-patterns write-up; the other three were in notes that are deleted when this PR lands. Nothing was lost, because that write-up already carried all four clusters as sections, in more detail than the plan had - the measured counter table, the eager-versus-cooperative reasoning. What it did NOT carry is the observation the plan could not have made either, so that is added: clusters 3 and 4 were the same defect twice. Each kept a local copy of a fact something else already owned, and the fix in both was to delete the copy rather than correct it - the load gate derived by conservation, the pause read from Kafka. The tell, stated for next time, is a field whose bugs are all of the form "X changed and Y did not": that is a second source of truth asking to be removed, not an accounting error to be balanced. Three citations repaired. Two quote or lean on the plan's own text, so they point at history (`git show a0ade8a:...`) as docs/citations.md prescribes, rather than being cut. THE A/B DEADLOCK PROBE MOVES TO #405 Same reasoning as the torture harness it now sits beside: an overnight instrument, run by hand, in this PR only because it was written here. Its A/B results stay - those are a finding about the deadlock, not a property of the instrument. That PR also gains a note on why neither instrument runs. The short version is that the obvious reading - "a scheduled soak lane needs building" - is wrong: experiments.yml already is one, and the real gap is that an A/B needs a control tree the lane cannot pass, and an eight-hour design does not fit a four-hour cap. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01UH2q829FDKQcqQseE8Cp7q
astubbs
added a commit
that referenced
this pull request
Sep 2, 2026
Ten commits, no Java overlap - master's core work (#257's produce-lock handover, #393's thread-confined consumer) is nowhere near the offsets package this branch changes. All three conflicts were in docs, and all three were two ledgers being appended to from both sides on the same day. WHAT THE RESOLUTION KEPT Both flake ledgers gained entries from master AND from this branch on 2026-09-01, so every conflict resolution keeps both sides rather than choosing one: - test-load-tightness-flakes.md: master's inFlightMessagesCommittedIfProcessed- DuringShutdown[1] paragraph and this branch's committedOffsetRemoved[1] latest recurrence are about different tests and neither supersedes the other. - test-untracked-ci-flakes.md: both sides added a processInKeyOrder row for sightings on the same day from different branches. Merged into one entry rather than two, because two rows under one test name is exactly the conflation master's own section warns against. This branch's four sightings are folded in as a sub-section of master's [3] bullet, where they turn "seen once, not reproduced" into a control-armed contention finding - and master's isolated-run result stops reading as a contradiction, since isolation removing the failure is what the contention reading predicts. - refactoring.md auto-merged: master deleted the produce-lock double-release OPEN QUESTION entry because #257 answered it. TWO REPAIRS MADE WHILE IN THERE Master's table had a blank line between two rows, which markdown renders as two tables with the second missing its header. Closed up. The JStream entry said "delete this entry when #116 lands" - a marker docs/inflight/AGENTS.md explicitly forbids on master, because the merge is the moment nobody is looking here. Replaced with the migrate-rather-than-delete outcome that doc names, and the attestation block it opened is now closed rather than running into the next section's. INHERITED AND READ #403 makes Node the default for new bin/ scripts and deletes check-shell-sigpipe.sh; #382 rewrites the hook guards to derive identity from the command rather than the session; a new core AGENTS.md binds @GuardedBy to any field change in the engine. This branch adds no scripts and changes no shared field, so none of the three change what it does - recorded because a green build is not evidence the ground under a design held still. AND ONE FIX THE MERGE COULD NOT BE COMMITTED WITHOUT #403's new bin/check-source-patterns.mjs refused this merge commit, reporting seven of #381's experiment runners as new shell scripts and advising --no-verify. They are master's, already grandfathered there, and this branch never touched them. The cause is that its `added-files` scope is the union of committed and staged additions against the merge base - and while a merge is staged but not yet committed, HEAD is still the pre-merge tip, so every file master added since the branch was cut is an addition against that base. It corrects itself the moment the merge commit exists, which is why nothing on master has seen it: it fires only in the window where the fix has to be applied. Membership of origin/master settles it without needing to know about merges at all - a path already there is not new to this repository, whoever staged it - so the subtraction lives in bin/lib/added-files.mjs as a pure function of three lists. That placement is what lets it be tested by control pair like every rule in the table, instead of standing up a fixture repository, which is the reason the runner had no test to break. The four must-NOT-match cases are red-proven against the pre-fix implementation. Carried here rather than split out because the alternative was committing this merge with --no-verify, and a gate that fires on somebody else's merged work teaches exactly that bypass.
astubbs
added a commit
that referenced
this pull request
Sep 7, 2026
… contract does not cover The commit before this one made "only the controller thread may call these" a declaration checked by ArchUnit reachability. That check is static, and its own javadoc enumerates three ways a poll-thread call can still arrive at a waiting acquire with the rule green: a reference that is STORED rather than invoked (ArchUnit's model gives a gauge or an executor task the same shape as a stream stage, so the walk cannot say WHEN a reach happens), a user-supplied `ConsumerRebalanceListener` (dynamic dispatch through an interface is out of reach of any deny list, and it is code the walk cannot start from), and a `synchronized` BLOCK (a MONITORENTER is not an access - the reason the rule would not have caught confluentinc#857 itself). None of those is closable by widening the rule. The note describes the guard that closes them, and it is NOT implemented here: the #393 ownership shape applied to `RetryQueue`. That PR's pieces are named so the next reader copies rather than designs - `ConsumerOffsetCommitter`'s volatile `Optional<Thread> owningThread`, set once by `claim()` when the owning thread's loop starts and read by `isOwner()`, and `PCModule.consumerManager()`'s thread-confinement wrapper, whose own comment states the lifecycle rule this would follow. Applied here: the controller claims the queue when the control loop starts (a lifecycle step, not a constructor argument, because the object exists before the thread that will own it), the `@ControllerThreadOnly` methods assert the current thread is the owner and fail loudly instead of taking the lock, and the guard is UNARMED while nobody has claimed - which is what leaves unit tests that drive a `RetryQueue` directly, and init-time use, unaffected. Cost: one reference compare per call, on a path already taking a lock. It is positioned against the convention on master rather than beside it. `parallel-consumer-core/src/main/java/bz/stub/parallelconsumer/AGENTS.md` owns "Declare thread confinement with `@ThreadConfined`, and assert it at the entry point", and the note cites it: the two existing patterns are `RetryQueue.RetryQueueIterator` (`@ThreadConfined(ThreadConfined.ANY)` plus `assertOnOwningThread`, with `RetryQueueIteratorConfinementTest` failing when the two disagree) and `ThreadConfinedConsumer`, both from #433. This guard is that shape with a NAMED thread - the control thread - which the rule says is the right value when the code really does pin one, and which is what gives the assertion something specific to compare against. Two design questions are what defer it, and both are stated rather than guessed. What "fail loudly" does inside `consumer.poll()`: a throw leaves a rebalance callback abnormally and its blast radius is the group, while "log an error and decline" is the behaviour the queue-first sweep already treats as safe - the better signal in a test is the worse one in production. And claim/release across a restart: `claim()` is called once and never released, so a controller that stops and starts, or a second processor in one JVM, needs a decision on whether a claim can be replaced, refused or dropped at close. Filed per `docs/inflight/AGENTS.md`: `core-` because it is product behaviour and the prefix names an area, `task` + `reliability` because no single defect is named yet, `concurrency` label, and `deferred` with what it waits on. No counts and nothing a command could answer. Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_018rEzrWYFr6oEzy6porczd3
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.
Extracted from #29 and cut fresh from master, where it compiles and passes
on its own. That PR's own decomposition plan called for this split and said why: this is the largest
piece of it by main-code volume, and its shape is a design decision about consumer ownership rather
than a patch, so it earns review separate from the lock fix it has been travelling with.
Description
KafkaConsumeris not thread safe, and this codebase runs two long-lived threads near it.ThreadConfinedConsumerwraps the consumer and refuses any call from a thread that does not own it.ConsumerOwnershipmakes that a lifecycle rather than a one-shot flag: the poll thread claims atstart of life, releases when the poll loop exits, and a closing thread may claim only what is
unclaimed or already its own.
Never stealing is the point. A guard that let close bypass the check would legalise closing
mid-poll, which is the failure it exists to prevent.
What ships here is the guard INSTALLED, not the guard GUARDING
Say this plainly, because the paragraphs above read as though the wrapper is already refusing calls,
and on this branch it is not.
AbstractParallelEoSStreamProcessorstill holds and calls the raw consumer fromoptions.getConsumer(), and nothing callsclaimConsumerOwnership(). Ownership therefore neverleaves
UNCLAIMED, whose documented meaning is that every thread is allowed - so no call isrefused at runtime, on any path, today. What this PR delivers is the wrapper, the ownership
lifecycle, their tests, and a seam that is correct and ready.
That is deliberate and it matches where the enforcement lives: the
ArchitectureTestrule thatpins the invariant stays on #29, and the claim call is wired there. It is
recorded here so a reviewer does not approve this as a shipped protection, and so nobody later reads
the unwired seam as an oversight and removes it.
Why this is separate from the deadlock fix
The measured claim on #29 is about roughly twenty lines of lock change - a
revoke path that declines with
tryLock()instead of blocking - verified by a four-cell controlacross both rebalance assignors. That control never measured this code. Reviewed together, the
verification of one invites being read as evidence for the other. They are independent in fact: the
deadlock fix references this wrapper only in comments and compiles without it.
It carries a regression it caused, and the test that proves it
Restructuring
ConsumerManagerfor thread confinement replaced a per-pollupdateCache()at pollentry with one-time priming in
init(). That left the pause cache written only when a pollended - so during every paused long poll, the control thread's back-pressure wakeup read the
previous poll's "not paused" value, never fired, and each pause cost a full long-poll timeout with
the pipeline drained. Measured as a four- to tenfold transactional throughput collapse that read as a
flaky CI lane for weeks.
The entry refresh is restored here, deliberately before
pollingBroker.set(true)rather thanafter it as on master, so
wakeup()cannot reach the consumer while those calls run - preserving thethread-confinement property this change exists for.
ConsumerManagerPauseCacheTestis red-proven on this branch, not assumed. Removing the entryrefresh fails it with
expected: 1; restoring it passes.MockConsumer.schedulePollTaskobservesthe cache from inside the poll it describes, so the test is deterministic - no timeouts, no
concurrency.
That the fix ships beside the code that caused it is the argument for this split rather than an
awkwardness of it.
Verification
mainandtest-compilecleanbin/check-all.sh- every gate passesWhat deliberately did not come with it
ArchitectureTeststays on fix(core) confluentinc#857: the revoke path declines the commit lock instead of blocking on it #29. It holds this work'sconsumer-holder allowlist and that branch's rebalance-blocking rule; splitting a shared file
drags unrelated work whichever way it is cut.
that branch; following the citation pulled in a plan, then two inflight notes, then
bin/scripts.A cascade that does not terminate is the tell that those documents are that PR's interconnected
research trail rather than this change's rationale. The citation now names the topic instead of a
path, because a path that does not resolve on the branch it is read from is worse than no citation.
The gate's escape hatch was not used.
Two things arrived after review, and one of them is unrelated to the subject
Answering the review at the code. The automated review read the two
ConsumerManager.subscribeoverloads as dead code. It was right that nothing calls them and right that nothing at the code
said so - the "guard INSTALLED" section above is where that fact lived, and a PR body is not where
the next reader of
ConsumerManagerlooks. The review reaching exactly the conclusion that sectionexists to prevent is the evidence a body-only note was not enough, so the reasoning now sits on the
methods. The review's second finding, that
init()'s javadoc offered an example its catch cannotactually see, was a real inaccuracy and is corrected. Both are comments only.
A SpotBugs finding on a line this branch wrote, which the review did not reach:
bin/check-pr-analysis-surfaces.shreportedHES_EXECUTOR_OVERWRITTEN_WITHOUT_SHUTDOWNon theforeignThreadassignment inThreadConfinedConsumerTestBase. The static lane is non-gating, so itwas green and silent. The fixture now shuts down what it is about to drop, chosen over the
rule-registry route because suppressing by rule would switch
HESoff repo-wide. That section ofthe report now reads
none.Unrelated to this change, and deliberately here anyway: the Unit Tests lane went red once on
ReactorBatchTest.simpleBatchTestunder KEY ordering - a known, already-tracked, still-undiagnosedflake, green on re-run. It is recorded as a fourth sighting in
docs/inflight/test-untracked-ci-flakes.md, which is the one file in this diff that has nothing todo with consumer ownership. It is here because that ledger's rule is to record a sighting before the
CI log carrying its evidence expires, and this one's evidence is worth keeping: the failing batch
payload shows the same three-way key collision as the 2026-08-25 sighting, in a different module,
which promotes the ledger's untested expectation-versus-input reading to the leading one.
Checklist
all carry their reasoning in javadoc at the point of use
docs/features/ThreadConfinedConsumerTest,ConsumerManagerCloseOwnershipTest,ConsumerManagerPauseCacheTest(red-proven), andConsumerManagerCommitRetryBudgetTestupdated for the wrapped constructor
@claude review thison the PR instead, which is the route that can openinline review threads