fix(offsets): an offset map claiming offsets the partition does not hold is refused at the first batch

docs/features/invalid-offset-metadata-policy.yaml

@@ -28,7 +28,9 @@ opt_in:
   isolated_from_stable_modules: false
 boundaries:
   - The policy governs every payload this build cannot read - a magic byte no encoding claims, an encoding with no decoder here, metadata recognized as Kafka Streams', and a body that is not something any encoder here could have produced.
-  - It does not govern a payload that decodes successfully into a wrong-but-plausible offset map; nothing in such a payload proves it wrong.
+  - A decoded map claiming offsets the partition does not hold is governed too: the first batch of records for a partition carries where its log ended when it was fetched, and a map reaching past the last offset the partition holds is treated as unreadable before any record is checked against it.
+  - No broker request is made for this: the watermark comes from the fetch that delivered the records. Until one has, the check defers to the next batch rather than judging a map it has no ground truth for.
+  - It still does not govern a payload that decodes into a map which is merely wrong about offsets that do exist; nothing in such a payload proves it wrong.
   - Under IGNORE the offset map is discarded, not repaired, so records completed but not yet committed before that point are replayed.
 references:
   - label: Offset-metadata policy

docs/inflight-tool.md

@@ -331,6 +331,7 @@ nearly every file. Each row carries the cheap signals a script can see:
           all-cited-numbers-settled: cites one fork number and it is settled: astubbs#207 MERGED (pull-request)
           bin/inflight.mjs docs show docs/inflight/bug-run-length-plausibility-ceiling.md
 ```
+<!-- file-refs: N/A - sample vet output recorded on 2026-09-07; the note it shows was retired by astubbs#480, and the sample is kept as the shape of the output, not as a citation -->
 
 **A signal is a reason to open the note, never a verdict.** That note may cite astubbs#207 because
 that pull request is where the defect was found, or because it fixed it - the row cannot tell, and

docs/inflight/bug-a-huge-incomplete-run-allocates-before-anything-can-check-it.md

@@ -0,0 +1,28 @@
+# A run-length payload can ask for two billion boxed longs before any check runs
+
+<!-- inflight-type: bug -->
+<!-- inflight-impact: crash -->
+
+`OffsetRunLength.runLengthDecodeToIncompletes` walks an *incomplete* run one offset at a time into a
+`TreeSet`, so a five-byte payload whose first run is `Integer.MAX_VALUE` asks for two billion boxed
+longs and exhausts the heap during the rebalance callback. Same family as the `BitSetV2` case
+astubbs#207 closed by requiring the declared bit length to be backed by bytes that are present - but
+run-length is compact by design, so nothing in the payload bounds what one entry may demand.
+
+**The plausibility check does not cover this one, by construction.** That check compares the
+*decoded* claim against the partition's log end offset, and the log end offset is not knowable until
+the first fetch delivers it - by which time this decode has already happened. The claim-side defect
+(a completed run marking a range as done, so PC silently skips it) is what it closes.
+
+**A decode-side bound would need a number nobody can derive**, which is exactly what
+`bound-a-guard-with-ground-truth-not-a-plausible-number-2026-09-08.md` argues against for the claim
+case. Two shapes worth considering when this is picked up, neither costed yet:
+
+- Cap the *materialisation* rather than the claim: decode incompletes lazily, or refuse to build a
+  set larger than the partition could deliver in flight, which is a memory-safety limit rather than
+  a plausibility one and can be stated as such.
+- Carry the claim without expanding it, so a range only becomes a set when records arrive to justify
+  it.
+
+Reachable by anything that can write offset metadata for the consumer group. It fails loudly (an
+`OutOfMemoryError`, not a silent skip), which is why it is not a v6 gate on its own.

docs/inflight/pr-207-offset-encoding-policy.md

@@ -151,8 +151,6 @@ Each of these outlived this PR and now has its own note, so nothing is restated
 
 - [`core-bytearray-encodings-have-no-codec.md`](core-bytearray-encodings-have-no-codec.md) - the two
   encoder-less, decoder-less `OffsetEncoding` constants, and the wire-format decision about them.
-- [`bug-run-length-plausibility-ceiling.md`](bug-run-length-plausibility-ceiling.md) - a structurally
-  valid but implausibly large run length, which nothing in the payload proves wrong.
 - [`core-fail-policy-escapes-the-rebalance-callback.md`](core-fail-policy-escapes-the-rebalance-callback.md)
   - `FAIL` stopping via Kafka's generic callback wrapper rather than PC's own fatal path.
 

docs/solutions/architecture-patterns/bound-a-guard-with-ground-truth-not-a-plausible-number-2026-09-08.md

@@ -0,0 +1,154 @@
+---
+title: "Bound a guard with ground truth, not a plausible number - the run-length ceiling that stayed open for a year"
+date: 2026-09-08
+category: architecture-patterns
+module: parallel-consumer-core/offsets
+problem_type: architecture_pattern
+component: offset_encoding
+root_cause: the_only_available_bound_was_a_guess_so_the_guard_was_never_written
+resolution_type: code_fix
+severity: high
+applies_when:
+  - A validator must reject an input that is structurally legal but semantically impossible
+  - The obvious bound is a constant somebody has to choose, and choosing it wrong rejects real data
+  - A guard has been parked as "a product call" because nobody could justify the number
+  - Deciding whether a check may make a network call, and what it should do when that call fails
+  - A validator wants a fact the process does not have yet, but will have shortly, for free
+related_components:
+  - PartitionState
+  - ConsumerManager
+  - EpochAndRecordsMap
+  - OffsetRunLength
+  - EncodedOffsetPair
+tags:
+  - guard-predicate
+  - ground-truth
+  - fail-open
+  - lazy-validation
+  - rebalance-path
+  - offset-encoding
+  - input-validation
+  - wire-format
+  - issue-197
+---
+
+# Bound a guard with ground truth, not a plausible number
+
+## Context
+
+PC stores its offset map in the commit metadata, and one encoding is run-length: a list of counts of
+completed and incomplete offsets. astubbs#207 hardened that decoder against everything the *bytes*
+can settle - a negative run, a body that is not a whole number of entries, a declared bit length with
+no bytes behind it - and routed every rejection through the user's `invalidOffsetMetadataPolicy`.
+
+One case was left, and recorded as parked: a run length of `Integer.MAX_VALUE` is structurally
+perfect. It moves the highest-seen offset about two billion forward, and `PartitionState`'s
+`isRecordPreviouslyCompleted` then reads every real record in that range as already succeeded. Not
+replay - **silent non-processing**, from metadata anything sharing the consumer group can write.
+
+The note that tracked it said, correctly, that a long run of completed offsets is exactly what
+run-length encoding is *for*, so nothing in the payload proves an absurd one wrong; and it concluded
+that bounding it "means choosing a plausibility ceiling - a number with no principled derivation",
+which is a product call rather than a correctness one. That framing is what kept the defect open: it
+treats "pick a constant" as the only available shape for the guard.
+
+## Guidance
+
+**When a check needs a bound, ask what in the system already knows the answer before you invent a
+number.** Three candidate bounds were considered here, and only one is not a guess:
+
+- **A configured window** - concurrency, the in-flight target, "nobody legitimately gets this far
+  ahead". Refuted by the product itself: PC's back-pressure keys off the *encoded payload size*, and
+  a run-length map with one stuck offset stays three entries wide however far the partition runs
+  ahead of it. A single poison record therefore produces a legitimately enormous range, so any such
+  ceiling eventually discards a true offset map - and discarding one replays every record it covers.
+- **A round number** ("no partition legitimately runs a billion ahead of its commit"). The same
+  objection with less honesty about it, and it fails silently: the day it is wrong, a real map is
+  thrown away and nothing distinguishes that from the corrupt case it was written for.
+- **The partition's log end offset** - the one bound a legitimate map provably cannot cross. PC only
+  ever encodes offsets it has polled, an offset the partition does not hold cannot have been polled,
+  and Kafka's end offset only grows, so a bound read *now* still holds for a map written earlier.
+  There is no false-positive case to trade off, which is what makes it a correctness bound rather
+  than a tuning parameter.
+
+**Having found the authority, ask what it costs to consult - and whether the answer is already in
+the building.** The first implementation of this fix bought the end offset with a `ListOffsets`
+request at assignment, batched across the partitions carrying metadata. It worked, and it was
+rejected on review for two reasons worth keeping:
+
+- **It put a blocking broker round trip inside the rebalance callback**, which in this codebase is
+  the most fragile path there is - the one whose stalls and deadlocks have consumed more
+  investigation than anything else. A check against corrupt metadata is not worth a new way for a
+  rebalance to hang.
+- **It failed open exactly when the broker was unhealthy.** An unreachable leader means no end
+  offset, which means the guard stands down - so the check was reliably absent in precisely the
+  conditions that make rebalances storm and offset maps get rewritten.
+
+**The same fact was already arriving for free, slightly later.** Every fetch response carries the
+partition's high watermark, and the consumer exposes it as `Consumer.currentLag` (KIP-695) - no
+request of its own, computed from the last fetch. `position + currentLag` reconstructs the watermark
+exactly when both are read in one breath on the thread that owns the consumer. So the check moved
+off the wire and onto the first batch: the decode accepts the map and *retains what it claims*, and
+the claim is settled before any record is consulted against it. Nothing is lost by waiting, because
+nothing consults the map until a record arrives to be consulted about.
+
+**"Later, for free" beats "now, at a cost" whenever the deadline is not real.** The instinct is to
+validate at the point of parsing, and here that instinct is what forced the wire call: the data
+needed to judge the payload simply is not in the process yet at decode time. Ask when the judgement
+is actually *needed* - the first consult, not the parse - and the cost can disappear.
+
+**A guard with no ground truth must stand down, not guess - and standing down is deferral, not
+acceptance.** `currentLag` is empty until a fetch has happened, and the position read is asked for
+with a zero timeout so it raises rather than sends a request. Every such outcome means *not
+established*, and the check simply looks again at the next batch. The batch's own records serve as a
+floor in the meantime - a record that arrived certainly exists - and a floor can only ever *confirm*
+a claim, never refute one, which is why it is safe to use with no watermark at all. Failing closed
+would discard honest offset maps on a broker hiccup, a far more likely event than the corrupt
+payload the guard exists for.
+
+**Check the producing side, and pin it at the boundary.** A bound like this is only safe if the
+encoder can never write what it rejects. Here it cannot, because the encoder's range top is an
+offset PC actually saw - but the argument is worth a test rather than a paragraph: every encoder
+round-trip case is measured against the rule with the tightest honest bound there is, a partition
+ending exactly on the last offset encoded. Anything that made the encoder run one offset past what
+it saw surfaces as a false rejection instead of as silence.
+
+**A check moved to where the meaning lives often covers more than the one it replaced.** The
+decode-side version had to be threaded into each decoder and applied per encoding; the lazy one
+reads the *decoded claim*, so run-length and bitset are covered by the same line, and any encoding
+added later is covered without being told. The bitset case is worth naming because it is easy to
+dismiss: its declared bit count must be backed by bytes that are present, so a full 4KB metadata
+field buys tens of thousands of skipped records rather than two billion. Bounded is not true.
+
+## Why This Matters
+
+The defect was correctly diagnosed, correctly filed, and then sat open - not because it was hard to
+fix but because the fix had been framed as a number nobody could justify. That framing is the durable
+lesson: **"this needs a constant we cannot derive" is a signal to go looking for an authority, not a
+reason to park the work.** The authority here was one method call away on an object the code already
+held.
+
+It also inverts the usual reason a guard is refused. The objection to bounding a run length was that
+the check might be wrong; the check that was finally written *cannot* be wrong in the expensive
+direction, and the only thing traded away is a round trip on a path that already makes several.
+
+## When to Apply
+
+- Writing any validator for a value whose legality depends on something outside the payload: a
+  length, an offset, a count, a timestamp, an id range.
+- Reviewing a parked bug whose blocker is "we would have to pick a threshold".
+- Deciding what a guard does when the fact it needs is unavailable - name the direction it fails in
+  and say why that direction is the cheaper wrong answer.
+- Being tempted to buy a fact with a network call on a latency- or liveness-critical path: ask what
+  already arrives carrying it, and when the answer is first *needed* rather than first available.
+
+## Related
+
+- [a-guard-must-assert-what-it-means-not-what-is-easy-to-check.md](a-guard-must-assert-what-it-means-not-what-is-easy-to-check.md) -
+  the sibling rule, and the step before this one: this document assumes you already know what the
+  guard should mean, and is about where the number it compares against comes from.
+- [`docs/inflight/pr-207-offset-encoding-policy.md`](../../inflight/pr-207-offset-encoding-policy.md) -
+  the policy this guard routes through, why the default is `IGNORE`, and the corrupt-payload family
+  it already covers.
+- [`docs/inflight/bug-no-metric-for-discarded-offset-metadata.md`](../../inflight/bug-no-metric-for-discarded-offset-metadata.md) -
+  still open: every discard here is logged and nothing counts it.

parallel-consumer-core/src/main/java/bz/stub/parallelconsumer/internal/BrokerPollSystem.java

@@ -124,10 +124,10 @@
         // See docs/solutions/test-flakiness/pc-silent-stall-under-contention-2026-07-29.md
         consumerMgr.setCloseInProgressSignal(this::isCloseInProgress);
 
         switch (options.getCommitMode()) {
             case PERIODIC_CONSUMER_SYNC, PERIODIC_CONSUMER_ASYNCHRONOUS -> {
                 ConsumerOffsetCommitter<K, V> consumerCommitter = new ConsumerOffsetCommitter<>(consumerMgr, wm, options);
                 committer = Optional.of(consumerCommitter);
             }
         }
         pcMetrics = pc.getModule().pcMetrics();
@@ -198,12 +198,12 @@
 
                 maybeDoCommit();
 
                 switch (runState) {
                     case DRAINING -> {
                         doPause();
                     }
                     case CLOSING -> {
                         doClose();
                     }
                 }
             }
@@ -331,8 +331,11 @@
 
         log.debug("Poll completed");
 
-        // build records map
-        return new EpochAndRecordsMap<>(poll, wm.getPm());
+        // Build the records map, and take the fetch's own high watermark with it while we are still on the thread
+        // that owns the consumer - see ConsumerManager#logEndOffsetIfKnownWithoutBlocking. The control thread cannot
+        // ask this question later: the consumer is confined here, and by then the answer would belong to a different
+        // fetch.
+        return new EpochAndRecordsMap<>(poll, wm.getPm(), consumerManager::logEndOffsetIfKnownWithoutBlocking);
     }
 
     private void checkStateForPausingSubscriptions() {

parallel-consumer-core/src/main/java/bz/stub/parallelconsumer/internal/ConsumerManager.java

@@ -22,6 +22,7 @@
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.OptionalLong;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
@@ -178,7 +179,7 @@
                 pendingRequests.addAndGet(-1L);
             }
         } catch (WakeupException w) {
             correctPollWakeups++;
             log.debug("Awoken from broker poll");
             log.trace("Wakeup caller is:", w);
             records = new ConsumerRecords<>(UniMaps.of());
@@ -231,7 +232,7 @@
     public void commitSync(final Map<TopicPartition, OffsetAndMetadata> offsetsToSend) {
         // we don't want to be woken up during a commit, only polls
         boolean inProgress = true;
         noWakeups++;
         Instant startedTime = Instant.now();
         while (inProgress) {
             try {
@@ -326,7 +327,7 @@
                 inProgress = false;
             } catch (WakeupException w) {
                 log.debug("Got woken up, retry. errors: " + erroneousWakups + " none: " + noWakeups + " correct:" + correctPollWakeups, w);
                 erroneousWakups++;
             } finally {
                 pendingRequests.addAndGet(-1L);
             }
@@ -377,14 +378,14 @@
     public void commitAsync(Map<TopicPartition, OffsetAndMetadata> offsets, OffsetCommitCallback callback) {
         // we dont' want to be woken up during a commit, only polls
         boolean inProgress = true;
         noWakeups++;
         while (inProgress) {
             try {
                 consumer.commitAsync(offsets, callback);
                 inProgress = false;
             } catch (WakeupException w) {
                 log.debug("Got woken up, retry. errors: " + erroneousWakups + " none: " + noWakeups + " correct:" + correctPollWakeups, w);
                 erroneousWakups++;
             }
         }
     }
@@ -494,6 +495,41 @@
     }
 
     public void onCommitRequested() {
         this.commitRequested = true;
     }
+
+    /**
+     * Where the partition ends (exclusive), <b>if the consumer can answer out of what the last fetch already told
+     * it</b> - the high watermark that came back with the records, at no cost and with no request of its own.
+     * <p>
+     * {@link Consumer#currentLag(TopicPartition)} (KIP-695, kafka-clients 3.0+) is the high watermark minus the
+     * current position, computed from the last fetch response, so {@code position + lag} reconstructs the watermark
+     * exactly - both read here in one breath, on the thread that owns the consumer, so they cannot describe
+     * different moments.
+     * <p>
+     * <b>Nothing here may block or throw.</b> The position is asked for with a zero timeout, so a partition with no
+     * position yet raises {@link TimeoutException} instead of sending an {@code OffsetFetch}; the lag is empty until
+     * a fetch has happened. Every such outcome is the same answer - <em>not established</em> - and the caller's
+     * contract (see {@code PartitionState#maybeVerifyLoadedOffsetMapAgainstThePartition}) is to try again on the
+     * next batch rather than to assume anything. An offset-map check must never be able to slow a poll down, let
+     * alone fail one.
+     *
+     * @return the log end offset, exclusive, or empty when it is not knowable without asking the broker
+     */
+    public OptionalLong logEndOffsetIfKnownWithoutBlocking(TopicPartition partition) {
+        try {
+            OptionalLong lag = consumer.currentLag(partition);
+            if (!lag.isPresent()) {
+                return OptionalLong.empty();
+            }
+            long position = consumer.position(partition, Duration.ZERO);
+            return OptionalLong.of(position + lag.getAsLong());
+        } catch (Exception e) {
+            // Includes TimeoutException (no position without a request), IllegalStateException (the partition went
+            // away under a rebalance) and any consumer implementation that does not answer this question at all.
+            log.debug("Consumer cannot say where {} ends without blocking, so nothing is checked against it this " +
+                    "time round", partition, e);
+            return OptionalLong.empty();
+        }
+    }
 }