Merge branch 'feat/225-uncommitted-completions-ledger' into docs/225-producer-fencing-brainstorm

This PR now stacks on the ledger rung (#474), itself on the groundwork rung
(#472), both cut out of this branch's tree today. Resolved by hand: the
processor and the module resolve to this branch's versions, which are the rungs' text plus recovery;
the README template takes the ledger rung's new retention section and points the recovery section's
retention sentence at it; a restore entry point the merge placed twice in WorkManager is kept once.
The ledger rung also carries two master commits this branch had not merged yet (the rebalance-callback
rule and the Lincheck shard harness), which arrive with it.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VrpH51xNDodaajE4P2nhFg

Author: astubbs

README.adoc

@@ -1102,14 +1102,22 @@ It is covered by an existing test, but no negative control was observed for it,
 The produce-lock timeout returning the record for retry started out the same way and no longer is: the eager-processing test holds the commit lock shut until a worker's acquisition is observed to time out, then watches that record's retry succeed, and removing the hold turns it red.
 It is recorded `PROVED`.
 
+[[completed-but-uncommitted]]
+=== What transactional mode keeps until the commit lands
+
+In the transactional commit mode, every record that completes is kept, with its offset, until the commit that carries it succeeds - on both producer paths.
+That is what makes it possible to put a transaction's work back into processing if the transaction is aborted rather than committed: the records are still there to re-register.
+The cost is memory held per partition that grows with `commitInterval`; the default interval keeps it to a fraction of a second of throughput.
+Producer recovery, below, is what uses the replay.
+
 [[producer-recovery]]
 === Producer recovery
 
 Where PC built the producer, a producer the broker reports invalid is replaced and the work its aborted transaction discarded is processed again, so processing continues instead of the instance stopping.
 The replacement is built from the same configuration, `transactional.id` included, and initialising it is what fences the producer it replaces; a finished `Producer` instance carries no configuration to rebuild from, so on that path these conditions keep their earlier behaviour.
 Recovery is a transactional-mode mechanism, because only a transactional producer has anything to recover from: in the consumer-commit modes PC's producer is non-transactional, so it cannot be fenced and has no epoch or generation to lose, an expired producer id is healed by the idempotent producer itself (KIP-360), and a send that still fails is retried record by record as it always was.
 What recovery adds is what only a transaction needs: abort, replay the completed-but-uncommitted work, rebuild under the same `transactional.id`.
-To make the replay possible, transactional mode keeps every completed record until the commit that carries it succeeds - on both producer paths - so memory held per partition grows with `commitInterval`; the default interval keeps it to a fraction of a second of throughput.
+The replay is possible because of what transactional mode keeps - see <<completed-but-uncommitted>>.
 The conditions PC treats this way are `ProducerFencedException`, `InvalidProducerEpochException`, `InvalidPidMappingException`, `OutOfOrderSequenceException` (and so `UnknownProducerIdException`) and `CommitFailedException`, on both the commit path and the produce path, unwrapped from the `ExecutionException` a send future raises and the `KafkaException` the client wraps a stored error in.
 Recovery repeats as often as the condition recurs; consecutive recoveries with no successful commit between them are paced by an exponential backoff, logged at ERROR, and visible on the `pc.producer.consecutive.recoveries` gauge, so an instance that is alive but not progressing is distinguishable from one that recovered once.
 A replacement that cannot be built yet is retried on a later cycle; one that can never be built - an authorization failure - ends the instance, naming the `transactional.id` that was refused.

docs/inflight/core-retry-queue-needs-a-runtime-controller-ownership-guard.md

@@ -0,0 +1,86 @@
+# The retry queue's waiting methods have a declared contract and no runtime guard
+
+<!-- inflight-type: task -->
+<!-- inflight-impact: reliability -->
+<!-- inflight-labels: concurrency -->
+<!-- inflight-state: deferred - the static half has landed; this half needs a decision on what "fail loudly" does inside consumer.poll() -->
+
+`RetryQueue.remove`, `add`, `removeAll` and `clear` take the write lock unconditionally, so a caller that is
+not allowed to wait must never reach them - which in this engine means the controller thread and nothing else.
+That contract is now DECLARED, by
+[`@ControllerThreadOnly`](../../parallel-consumer-core/src/main/java/bz/stub/parallelconsumer/state/ControllerThreadOnly.java),
+and CHECKED statically, by `ArchitectureTest.rebalanceCallbacksMustNotBlock`, which reports a reach into an
+annotated method exactly as it reports a reach into a deny-listed JDK blocking call. This note is the other
+half: the runtime guard that would hold the same contract where a static walk cannot follow.
+
+## The shape, copied from a guard this repo already runs
+
+astubbs/parallel-consumer#393 did it for the consumer, and the pattern is two pieces:
+
+- `ConsumerOffsetCommitter` holds a `volatile Optional<Thread> owningThread`, set once by `claim()` when the
+  poll thread's control loop starts, and read by `isOwner()` - `Thread.currentThread().equals(owningThread
+  .orElse(null))`. The claim is a LIFECYCLE step taken by the owning thread itself, not a constructor
+  argument, because the object is built before the thread that will own it exists.
+- `PCModule.consumerManager()` wraps the user's consumer in `ThreadConfinedConsumer` (grep
+  `thread-confinement`), whose own comment states the rule this note copies: ownership is claimed when the
+  loop starts, and calls before that are allowed from any thread.
+
+Applied to `RetryQueue`: the controller claims the queue when the control loop starts; every
+`@ControllerThreadOnly` method asserts that `Thread.currentThread()` is the owner and fails loudly instead of
+taking the lock; and the guard is UNARMED while no controller has claimed, so unit tests that drive a
+`RetryQueue` directly - and any init-time use - are unaffected. Unarmed-by-default is the part that makes this
+cheap to land: it changes nothing until a real control loop exists.
+
+## Which annotation, and why the pairing rule makes this note necessary
+
+`parallel-consumer-core/src/main/java/bz/stub/parallelconsumer/AGENTS.md` owns the rule - "Declare thread
+confinement with `@ThreadConfined`, and assert it at the entry point". Infer's `@ThreadConfined` is CONSUMED by
+RacerD and never checked, so an unpaired one silences a detector and is worse than no annotation at all. The
+two existing patterns are named there: `RetryQueue.RetryQueueIterator` carries `@ThreadConfined(ThreadConfined
+.ANY)` plus an `assertOnOwningThread` at every entry point with `RetryQueueIteratorConfinementTest` failing
+when the two disagree, and `ThreadConfinedConsumer` is the older hand-rolled version of the same idea for the
+poll thread. Both were established by astubbs/parallel-consumer#433.
+
+**The guard this note describes is that shape with a NAMED thread rather than `ANY`** - 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. `@ControllerThreadOnly` is deliberately not that annotation:
+no analyser reads it, so it silences nothing and the pairing rule's rationale does not reach it. When the
+runtime guard lands, the marker may fold into the `@ThreadConfined` + assertion pair.
+
+## What it covers that the static rule cannot
+
+The ArchUnit rule's own javadoc enumerates its blind spots, and each one is a way for a poll-thread call to
+arrive at a waiting acquire with the rule green:
+
+- **A stored reference.** ArchUnit's model gives a reference invoked now (a stream stage) the same shape as
+  one invoked later (a metrics gauge, an executor task), so the rule cannot say WHEN a reach happens - it is
+  conservative about immediate reaches and silent about deferred ones.
+- **A user-supplied `ConsumerRebalanceListener`.** Dynamic dispatch through an interface is out of reach of any
+  deny list, and a user listener is exactly the code the walk cannot start from.
+- **A `synchronized` block.** A `MONITORENTER` is not an access, so it is invisible at any depth - the reason
+  the rule would not have caught confluentinc#857 itself.
+
+A runtime owner check does not care how the call arrived. It costs one reference compare per call on a path
+that is already taking a lock.
+
+## Open design questions - these are what defer it
+
+- **What "fail loudly" means on the poll thread**, which is the binding one. The call is inside
+  `consumer.poll()`, so a thrown exception leaves a Kafka rebalance callback abnormally and its blast radius
+  is the group, not the caller; "log an error and decline the removal" is the alternative. Throwing is the
+  better signal in a test and the worse one in production, which is the trade to settle - and
+  astubbs/parallel-consumer#431 settles the same trade for the static half by declining rather than throwing,
+  which is the precedent to weigh rather than a decision already taken here.
+- **Claim and release across a restart.** `ConsumerOffsetCommitter.claim()` is called once and never released.
+  A controller that stops and starts again, or a second `ParallelEoSStreamProcessor` in the same JVM, needs a
+  decision on whether a claim can be replaced, refused, or dropped at close - and on what an assertion does in
+  the window between them.
+
+## Where to look when picking this up
+
+- `ConsumerOffsetCommitter`, grep `owningThread` and `isOwner` - the reference implementation, including why
+  the claim is invisible to a grep for `.claim(`.
+- `PCModule`, grep `thread-confinement` - where a wrapper is wired, and the "claimed when the loop starts"
+  rule stated in a comment.
+- `docs/inflight/static-archunit-main-code-rules.md`, "the rule now enforces a contract the CODEBASE declares"
+  - what the static half does and what it measured.

docs/inflight/core-stale-arrival-guard-needs-a-null-safety-decision.md

@@ -81,3 +81,25 @@ NullAway lane.
 Until both are answered, this stays parked. The mechanism it guards is documented in
 `PartitionState#epochIsStale`'s three-checkpoint javadoc, which also records why re-checking per
 record or consulting the live epoch closes nothing.
+
+## 2026-09-03: a fourth fixture, found by a harness rather than by the prototype
+
+Added beside the text above rather than over it - the decision is unchanged, its evidence is not.
+
+`ShardManagerLincheckTest` reached `ProcessingShard.isWorkContainerStale`'s unguarded deref of
+`PartitionStateManager.getPartitionState` and reported `NullPointerException` in CI. It is the same
+shape as the three fixtures named above - a harness driving a `PCModuleTestEnv` whose partition was
+never assigned - and it had simply never taken the branch before: the branch needs an arrival to find
+a RESIDENT at its offset, and while every revoke sweep removed the resident and, under KEY ordering,
+the empty shard with it, every add was an insertion into nothing. A sweep that DECLINES leaves the
+resident in place, so the next add takes the branch.
+
+**The sighting was on astubbs/parallel-consumer#431's branch, which is still open**, and the
+declining sweep is what that PR adds - so the reach is not yet available on `master`, and
+`ShardManagerLincheckTest` was re-measured green here without the fixture change. Recorded now rather
+than when that PR lands, because the datum is about this note's question and does not depend on its
+outcome: the unguarded call is reachable from a *state the product deliberately creates*, not only
+from a test shortcut, which is a datum for the second bullet under "The decision needed".
+
+Fixed as a fixture - the harness now assigns its partition, which is what its own constructor claims
+to model - so this note's policy question is untouched.

docs/inflight/static-archunit-main-code-rules.md

@@ -171,6 +171,56 @@ wait, owned by astubbs#44 - and grew on 2026-08-31 when the rule's deny list was
 defect-class sweep and immediately found a second, pre-existing defect on master
 ([`bug-retry-queue-write-lock-on-the-rebalance-path.md`](bug-retry-queue-write-lock-on-the-rebalance-path.md)).
 
+**Update 2026-09-07: the walk now follows METHOD REFERENCES, and the list grew by twelve because of it.**
+`notReachBlockingCalls()` followed `getMethodCallsFromSelf()` and nothing else, and ArchUnit models
+`retryQueue::remove` as a method REFERENCE, which that accessor never returns. So
+`ShardManager.removeStaleContainers` - `.map(retryQueue::remove)`, a real unbounded write-lock acquire - was
+invisible to the rule from every rebalance callback, including `onPartitionsAssigned`, which the list did not
+mention at all. `getMethodReferencesFromSelf()` is now walked beside the calls, through one shared
+`inspectReach()` that applies the same deny-list check, the same `root => target` exemption key, the same
+synchronized-method check and the same enqueue to both kinds.
+
+**This is the "a complete-looking allowlist is evidence about the WALK" case, and it is the third blind spot
+this rule has had.** The list read as finished while three callbacks sat on a waiting acquire; nothing was
+wrong with the entries, only with what could reach the point of being entered. Widening the walk with the
+entries untouched is what measured it - the rule went from green to eighteen violations - and the twelve new
+`root => target` keys that produced are in the list with astubbs/parallel-consumer#431 named as the owner that
+deletes them. Recording them rather than shipping the rule green is deliberate, and it is the argument below
+applied to itself: a gate that arrives green has measured nothing.
+
+**Update 2026-09-07: the rule also enforces a contract the CODEBASE declares, not only a JDK deny list.**
+`@ControllerThreadOnly`
+(`parallel-consumer-core/src/main/java/bz/stub/parallelconsumer/state/ControllerThreadOnly.java`) marks a
+method that may wait, and a reach into one is reported exactly as a deny-listed call is - same exemption key,
+same message shape, calls and method references alike. It closes a gap the deny list cannot: a
+`tryLock()`-based sibling of `RetryQueue.remove` would take the very same lock and be correctly absent from
+the list, so once both live on one class nothing but a declared contract can tell a waiting acquire from a
+declining one. astubbs/parallel-consumer#431 is the change that creates that pair.
+
+**It is deliberately NOT Infer's `@ThreadConfined`**, which arrived with astubbs/parallel-consumer#433 and is
+the subject of a rule in `parallel-consumer-core/src/main/java/bz/stub/parallelconsumer/AGENTS.md`. That one
+is CONSUMED by RacerD, so it must be paired with a runtime assertion or it silences a detector; this one is
+read by no analyser and silences nothing, so it is checked here and nowhere else. The runtime half - a
+named-thread `@ThreadConfined` plus an `assertOnOwningThread`, in the `RetryQueue.RetryQueueIterator` /
+`ThreadConfinedConsumer` shape - is tracked in
+[`core-retry-queue-needs-a-runtime-controller-ownership-guard.md`](core-retry-queue-needs-a-runtime-controller-ownership-guard.md).
+
+**The standing proof is a positive control, because a measurement in a commit message protects nothing.**
+`RebalanceCallbackRuleControlTest` hands the real rule object - not a copy - two hand-imported fixtures under
+`archfixture/`: one reaching a deny-listed acquire through a method reference, one reaching an annotated
+method that waits for nothing at all, so no entry in the deny list can match it. Both are hand-imported
+because the rule's own `@AnalyzeClasses` carries `DoNotIncludeTests`, which is also what keeps a deliberate
+violation from turning the production rule permanently red. Each half was checked by removal: drop the
+reference hop and both cases fail; drop the annotation block and the annotated case fails.
+
+**Constructor calls were the obvious next widening and were measured and rejected**, which is worth recording
+because it reads as free. Enqueuing `getConstructorCallsFromSelf()` turns every factory call into a reach into
+whatever the constructed object wires up: `PCModule.workManager()` contains `new WorkManager(..)`, whose
+constructor registers a metrics gauge as a method reference, and that gauge reads the retry queue under its
+read lock - a red rule on a path no callback takes. The general limit under it is that ArchUnit's model has
+the same shape for a reference invoked now (a stream stage) and one invoked later (a gauge, an executor task),
+so a reference-walking rule is conservative by construction, and cannot say WHEN a reach happens.
+
 **The tension is real and is not resolved here.** The argument for the entries: each names a defect
 that exists on master, has a tracking note, and was not introduced by the branch that had to decide
 what to do about it - the alternative was leaving a rule permanently red on inherited work, which