add metric for monitor max queued record num shard by flashmouse · Pull Request #905 · confluentinc/parallel-consumer

flashmouse · 2026-01-04T09:59:30Z

I want to add a metric to monitor the maximum number of queued records for the shard. This metric is especially useful for orderType.KEY. If this metric is high and all messages are consumed at the same speed, it implies that some shards may contain a hot key with lots of records.

Checklist

Documentation (if applicable)
Changelog

…onfluentinc#905) Cherry-pick of confluentinc#905 (author: flashmouse). Adds a SHARDS_MAX_SIZE gauge that reports the record count in the most-loaded shard. Useful with KEY ordering to detect hot-key bottlenecks. Also simplifies .keySet().size() to .size(). Upstream PR: confluentinc#905 Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Add a "Parallel-safe work while PR #57 is in flight" section to docs/inflight.md recording, for each in-flight track, whether it collides with PR #57's metrics/state files (857, 909, 51 -> sequence after) or is parallel-safe (912, release, logging cleanup, security bumps, contributor fixes, #40, confluentinc#915, DLQ), ranked by readiness. Also refresh the confluentinc#859 entry to the consolidated PR #57 (bundles the confluentinc#893/confluentinc#905 cherry-picks, supersedes the closed #42->#43->#45 stack) and expand the confluentinc#912 entry (ready, pushed, no PR, vertx-isolated). Bump the last-updated date. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

…onfluentinc#905) Cherry-pick of confluentinc#905 (author: flashmouse). Adds a SHARDS_MAX_SIZE gauge that reports the record count in the most-loaded shard. Useful with KEY ordering to detect hot-key bottlenecks. Also simplifies .keySet().size() to .size(). Upstream PR: confluentinc#905 Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Render backlink comments from an optional per-entry backlink field in upstream-map.yaml (source of truth) instead of a separate body, with the same {{FORK_REPO}}/{{FORK_REF}}/{{SUMMARY}}/{{ID}} placeholders; entries without it fall back to the generic templates. bug-859 uses it to explain the two-cause leak vs the already-merged upstream confluentinc#892. Make fork status honest about landed-ness. The old "fixed" conflated "fix written" with "shipped": every "fixed" entry is actually an OPEN, unmerged fork PR (or a branch with no PR). Replace with a lifecycle vocabulary (none|in-progress|ready|pr-open|merged|released|superseded|wontfix) and correct the entries: confluentinc#859/confluentinc#893/confluentinc#905 -> pr-open (in open PR #57), confluentinc#857 -> ready (branch-only). Add an optional per-entry todo: list for outstanding actions (merge the open PR, post the backlink) surfaced by "upstream-map.py todo", so "still to do" is explicit rather than implied by an open PR + null forwarded. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

…tus guard Audited PR #57 against issue confluentinc#859. The code matches the issue (registeredMeters List -> LinkedHashSet + prune, plus caching OffsetMapCodecManager in PartitionStateManager), but the wording framed the leak as rebalance-driven when the issue is commit-driven. Tighten bug-859 summary/notes/backlink: the List accumulated a duplicate Meter.Id on every registration (every commit); fork PR #57 fixes it via List->Set + PartitionStateManager caching (also closes #233); upstream confluentinc#892 covers the per-commit churn; confluentinc#893/confluentinc#905 are unrelated cherry-picks. Also fix upstream-backlink.sh: the fix-backlink status guard still checked the removed "fixed" value, so it refused every entry after the lifecycle-status change. Now allows ready|pr-open|merged|released and refuses none|in-progress. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

The description repeated itself ("queued records in the shards with the most queued records"). State it plainly: the maximum records queued in any single (most-loaded) shard. Upstream-PR: confluentinc#905 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

…onfluentinc#905) Cherry-pick of confluentinc#905 (author: flashmouse). Adds a SHARDS_MAX_SIZE gauge that reports the record count in the most-loaded shard. Useful with KEY ordering to detect hot-key bottlenecks. Also simplifies .keySet().size() to .size(). Upstream PR: confluentinc#905 Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

The description repeated itself ("queued records in the shards with the most queued records"). State it plainly: the maximum records queued in any single (most-loaded) shard. Upstream-PR: confluentinc#905 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Record the user-visible changes this PR introduces under the unreleased 0.6.0.0 section: the PCMetrics memory-leak fix, the accurate-committed-offset fix, and the new shards.max.size metric. Follows the fork/upstream reference convention. Upstream-Issue: confluentinc#859 Upstream-PR: confluentinc#893 Upstream-PR: confluentinc#905 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Add docs/runbooks/pr57-post-merge.md - the exact, reviewed upstream comments to post after #57 merges: the confluentinc#859 fix-backlink (from the manifest backlink field), and tailored author-crediting notes for the carried confluentinc#893/confluentinc#905 PRs (the generic template is wrong for cherry-picked PRs), plus the manifest status flips. Update AGENTS.md backlink guidance: always pre-draft backlinks as a runbook committed to the PR - reviewable in-diff, tailored per target, context-aware across targets - rather than running the backlink script blind. Runbooks live in docs/runbooks/, deleted once executed; stragglers swept at the next major release. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Cover the new max-queued-records-per-shard metric next to the existing SHARDS_SIZE assertion in PCMetricsTest - the larger of the two per-partition shard remainders. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

…ntinc#905 gauge perf Two review findings: static errorPolicy is set-once since #57, so multiple PC instances in one JVM with differing InvalidOffsetMetadataHandlingPolicy would all use the last-constructed policy (fix via instance field, part of the #233 de-static work); and the SHARDS_MAX_SIZE gauge re-walks shard queues, duplicating the SHARDS_SIZE traversal (negligible). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Add a Shards Max Size entry to the metrics reference in README_TEMPLATE.adoc (next to Shards Size) and regenerate README.adoc. The confluentinc#905 gauge was user-visible but undocumented in the metrics list. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Commit 8923603 resolved five conflicts with `git checkout --theirs`, which takes master ENTIRE file rather than the conflicting hunk. Where a file had changes from both sides, everything this branch had done in it went with it. Two casualties, both caught by the reviewer running the tests rather than reading the diff: ShardManager lost the confluentinc#905 SHARDS_MAX_SIZE gauge - the field and its registration - while PCMetricsDef.SHARDS_MAX_SIZE and the PCMetricsTest assertion on it merged cleanly and still expected it. PCMetricsTest.metricsRegisterBinding was failing with expected 785.0 but was -1.0, the no-such-gauge sentinel, and CHANGELOG.adoc still advertised the metric as delivered. The TODO(refactor) marker and the .size() cleanup went the same way. PartitionStateCommittedOffsetTest lost offsetToCommitIsComputedOncePerCommit, the confluentinc#893 regression test. The production fix was untouched, so nothing was functionally broken - but the test that would catch a regression back to the two-call dirty read was gone, which is the worse half of the two. Restored both, keeping master side where the sides genuinely differed: confluentinc#857 in ShardManager, and the tree-wide sweep wording in the test. Neither restore reintroduces the role form the gate now rejects. Verified by running them: PartitionStateCommittedOffsetTest 6/6, PCMetricsTest 2/2. The lesson is that --theirs and --ours are file-level operations, and a conflict is hunk-level. Resolve the hunk, or diff the pre-merge tip afterwards to see what the shortcut discarded. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QqHpNSXC39ANv9kG1ZvUzn

…e the rule's retelling A review round audited the `--theirs` incident this branch already wrote a rule about, and found the repair had been scoped to what the tests caught. The two casualties that broke tests were restored weeks ago; the ones that broke nothing were not. Restored: - The `shards.max.size` reference block in `src/docs/README_TEMPLATE.adoc` and the generated `README.adoc`. `CHANGELOG.adoc` has been advertising a metric the metrics reference did not document since merge `8923603bd` ate it - the same merge, resolved the same way, as the `confluentinc#905` gauge and the `confluentinc#893` test that were found. - Two `docs/refactoring.md` entries dropped by merge `44aadef19`, leaving two shipped `TODO(refactor)` comments ending "See docs/refactoring.md" pointing at sections that no longer existed. The third dropped item, the static `errorPolicy`, is genuinely obsolete - master made it an instance field - so it stays gone. Nothing fails when prose vanishes, which is why these survived a dozen review rounds: a merge that takes the other side renders as nothing at all, so there is no removal for diff-vs-base review to show. Also: - The three `CHANGELOG.adoc` entries go. `AGENTS.md` says a PR never adds to the changelog; the 0.6.0.0 section is regenerated from the commit log at release anyway, so they were inert as well as against policy. - The `docs/upstream.md` runbook rule goes. It told agents to draft from "the script's dry-run", but master deleted `scripts/upstream-backlink.sh` in `735b1d3ae`, and `9f4ae7b64` in this very branch deleted the runbook it mandates - mirroring the issues and letting `Fixes` close them leaves nothing to remember. - The shutdown-race note cited `BrokerPollSystem.java:278` for a swallow that is neither there nor a line number worth citing: `closeAndWait()` rethrows, and the `catch (Exception)` that only warns is at the call site in `AbstractParallelEoSStreamProcessor`. Re-anchored to the log line. - `PCMetrics859Test` called `registeredMeterCount()` package-private; it is public. The `--theirs` rule itself now states the rule and cites the incident instead of retelling it, which is what AGENTS.md asks of every rule and what it had stopped doing at 504 lines. The write-up gains what this audit added: audit every file the merge's conflict list names rather than stopping when the suite goes green, and normalise the namespace before diffing across a package rename, where the plain diff calls every file wholly rewritten and so says nothing. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012J2KjzpiUKg2tGFT7B2D2e

…ey is visible Adds the `shards.max.size` gauge (`PCMetricsDef.SHARDS_MAX_SIZE`): the number of records queued in the single most-loaded shard. `shards.size` already reports the total queued across all shards, which cannot distinguish work spread evenly from one key monopolising a shard. Under KEY ordering that difference is the thing an operator needs - a hot key serialises its whole shard while the totals look healthy. Reading the two gauges together makes the skew visible without adding per-shard cardinality to the registry. Cherry-pick of confluentinc#905 by flashmouse, whose original also simplified `processingShards.keySet().size()` to `.size()`; that is carried here unchanged. The gauge walks every shard queue to find the maximum, duplicating the traversal `SHARDS_SIZE` already performs, so each scrape walks the queues twice. That is negligible at present sizes and is recorded in docs/refactoring.md rather than optimised speculatively - the `TODO(refactor)` in `ShardManager` points at it. Documented in the metrics reference in src/docs/README_TEMPLATE.adoc, with README.adoc regenerated from it. Carries confluentinc#905. No fork issue exists for this metric: the upstream PR cites none, and the only related upstream issue, confluentinc#71 ("Health-checks"), is far too general to claim as its request. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012J2KjzpiUKg2tGFT7B2D2e

…anged Process learnings from this PR's own history, kept separate from the three fixes because they belong to none of them. **The `--theirs` incident.** Updating this branch from master, five files conflicted on a single comment line each and were resolved with `git checkout --theirs <file>`, which takes master's entire version. Two casualties broke tests and were repaired at the time: the confluentinc#905 gauge and the confluentinc#893 regression test. The repair stopped there, because that is where the tests stopped. A later audit found the `shards.max.size` README block and two `docs/refactoring.md` entries had gone the same way and stayed lost - nothing fails when prose vanishes, and a merge that takes the other side renders as nothing at all, so there is no removal for diff-vs-base review to show. Those are restored in the commits this one accompanies. AGENTS.md now states the rule and cites the write-up rather than retelling it, which is what that file asks of every rule and had stopped doing at 504 lines. The write-up adds what the audit established: audit every file the merge's conflict list names rather than stopping when the suite goes green, and normalise the namespace before diffing across a package rename, where the plain diff calls every file wholly rewritten and so says nothing. **The squash message rule.** docs/merge-checklist.md told agents to put the squash message "in the merge when you perform it, or in the PR body if the author is merging". The second half is wrong: a description tells reviewers what the change is and is read while the PR is open; a squash message is commit text consumed once, at merge. It is also not passive advice - a hook injects that file whenever a prompt looks like merge prep, so the instruction is re-asserted continuously. Corrected to name the merge itself, or handing the message over at merge time. **Two smaller corrections.** The test-class naming rule moved to docs/testing.md, its home since master's restructure. The copyright write-up drops a claim about `docs/inflight.md`, a file that no longer exists. `docs/todo-index.md` is regenerated, and `upstream-map.yaml` records confluentinc#894 against the confluentinc#893 cherry-pick, which the manifest had as `issues: []`. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012J2KjzpiUKg2tGFT7B2D2e

…r updated **`PartitionState.tryToEncodeOffsets()`'s javadoc described the old return type.** Master has `private Optional<String> tryToEncodeOffsets()`, where `@return if possible, the String encoded offset map` is accurate. This PR changed the return to `Tuple<Optional<String>, Long>` for confluentinc#893 - commit the offset the payload was encoded against, not a later one - and left the javadoc describing a type the method no longer returns. The replacement says why the two values travel together, since that pairing IS the fix: a caller that re-derives the offset reintroduces the defect. Nothing recorded this anywhere - not `docs/refactoring.md`, not the TODO index. It was a live inaccuracy in a signature this PR itself changed, not backlog. **`docs/inflight/pr-57-metrics-leak.md` recorded scope but nothing open.** `remind-inflight-on-push.sh` surfaces it on every push to this branch precisely so its contents can still land in this PR; it fired twice today and the note went unchanged. It now carries what is actually open, including two things a merger must not discover by accident: that the base is #325's branch rather than master, and a decision owed on the `SHARDS_MAX_SIZE` gauge. That gauge is worth stating in the log as well as the note. `ProcessingShard.entries` is a `ConcurrentSkipListMap`, whose `size()` traverses rather than being O(1), so `getCountOfWorkTracked()` is O(n). Master already pays it once for `SHARDS_SIZE`; the confluentinc#905 cherry-pick adds a second gauge walking the same structure, and Micrometer pulls gauges independently, so they cannot share a scan without a memoised snapshot. Each scrape is therefore O(total queued records), twice. The `shardEntryCounts()` helper added earlier in this PR did NOT change that - two traversals before it, two after; it only stopped the two expressions drifting apart. The `TODO(refactor)` in `ShardManager` carries the cheap fix, an O(1) counter maintained where `availableWorkContainerCnt` already is. Verified: core suite 398 pass / 8 skipped. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012J2KjzpiUKg2tGFT7B2D2e

… the shard gauges `docs/refactoring.md` has recorded this since before the PR: the confluentinc#905 `SHARDS_MAX_SIZE` gauge re-walks every shard queue, `getEntries().size()` is O(n) on a `ConcurrentSkipListMap`, so each scrape walks twice - and its verdict is "Negligible now; if it ever matters, derive both gauges from a single scan." I re-derived that from the code without checking, and wrote it into two more places, where it had already begun to drift: - The `TODO(refactor)` marker proposed a DIFFERENT fix - an O(1) counter on `ProcessingShard` - from the one the owning entry proposes. Two fixes for one problem, in two files, is how the next reader ends up implementing neither. - `docs/inflight/pr-57-metrics-leak.md` raised it as "a decision owed before merge", contradicting the owner's triage outright. A merge blocker invented out of an item already assessed as negligible is worse than not recording it: it spends the reviewer's attention on a settled question. The marker is now a pointer that names the owner and says not to restate the fix. The inflight bullet is gone. The owner entry gains the cross-link that would have prevented this: **Shard-count caching** in its own Performance section, carrying the upstream design draft `confluentinc#530` and the three abandoned branches that attempted it - which is the prior art the marker was reinventing. The prior-art check in AGENTS.md exists for exactly this and I skipped it; the entry was one grep away in the file I was already editing. Verified: core suite 398 pass / 8 skipped, todo-index regenerated, gates pass. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012J2KjzpiUKg2tGFT7B2D2e

…csRegisterBinding Folded in rather than kept as its own PR, because the two overlap directly and this is the soonest the test runs again. `PCMetricsTest.metricsRegisterBinding` is quarantined on master with a diagnosed mechanism, not a flake ledger: it asserted `PARTITION_LAST_COMMITTED_OFFSET` - a contiguous watermark bounded by the lowest incomplete offset - against a completion COUNTER, while the suite runs UNORDERED and workers block on a latch before incrementing. The gap was permanent by construction, so the 120s `atMost` only made it cost 140s of every CI run. The incoming commit freezes the assertions on offsets instead, and removes both the `@Quarantined` annotation and the registry entry. #57 is the right vehicle: it is already the PCMetrics PR, it owns `PCMetrics.java` and `PCMetricsDef.java`, and it carries `PCMetrics859Test` - so a test asserting metric semantics and the code defining those metrics would otherwise land in two different PRs. They also collide on `PCMetricsTest.java` and `docs/inflight/bug-857-family.md`, so kept apart one would have had to resolve this conflict anyway, later and with less context. **The conflict resolution is the part worth reading.** #57's contribution to that file was a 4-line exact assertion for the confluentinc#905 `SHARDS_MAX_SIZE` gauge, sitting inside the very method being rewritten - and rewritten precisely because exact equality against completion-derived comparands was wrong there. Ported into the new structure rather than re-applied verbatim: it now reads the same single counter snapshot inside `untilAsserted` that `SHARDS_SIZE` does, as `Math.max(quantityP0 - completedP0, quantityP1 - completedP1)`. Exact equality is correct for this one where it was wrong for the watermark, and the comment says why: a shard's depth genuinely IS quantity-minus-completed, with no contiguity requirement for a hole to defeat. Getting that distinction wrong would have reintroduced the flake class this merge exists to remove. Verified: `PCMetricsTest` runs unquarantined, 2 tests green in 10s against the 140s the quarantine entry recorded. Core suite 399 pass / 8 skipped - up one from 398, which is the test rejoining the run. All gates pass, including `check-quarantine-registry` and `check-quarantine-owners`. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_012J2KjzpiUKg2tGFT7B2D2e

…ey is visible Adds the shards.max.size gauge: the largest number of records queued in any single shard. shards.size already reported the total across all shards, which cannot distinguish an evenly loaded consumer from one where a single hot key is serialising all the work behind it - the case that actually hurts under KEY ordering. Cherry-pick of upstream confluentinc#905. Its test assertion travels in the un-quarantine commit rather than this one, because that commit rewrites the whole of PCMetricsTest and splitting the file between them would leave neither readable. Refs: confluentinc#905

…it on offsets The test was quarantined as flapping because it asserted PARTITION_LAST_COMMITTED_OFFSET against a shared completion COUNTER while the suite runs UNORDERED. Commits are contiguous and bounded by the lowest incomplete offset; completions are not ordered. Workers awaited a latch BEFORE incrementing, so a latched worker's offset never completed and the gap was permanent - the 120s atMost could not close it, only make the failure expensive. It passed only when the latched workers happened to hold the highest offsets, so a pass proved nothing. Rewritten to gate on each record's OWN offset rather than a shared counter, with two latched ranges held open per partition: a deliberate non-contiguous hole, and a freeze above a fixed offset. The pool is asserted wider than the workers the hole parks forever, because getting that wrong hangs rather than fails. Also carries the confluentinc#905 gauge assertion, since this commit rewrites the file wholesale and splitting it would leave neither half readable. Refs: #120, confluentinc#905

…d overshoot Three findings from the Codex review, all taken. **The silent-loss count was measured, printed, and never asserted.** `runGrowingPartition` counted `droppedWithoutEverRunning` - records dismissed against a fabricated `offsetHighestSucceeded` that no cycle ever ran - formatted it into the ledger, and returned only the worst overshoot. The two numbers are the defect's two mutually exclusive regimes: below the payload width the commit overshoots the log end, at or above it the overshoot stays exactly zero while records are lost. So every `K >= L` arm of both sweeps was green *by construction*, including the regime the write-up calls the stronger argument for taking the fix. The sweep now returns both numbers and `assertBothRegimes` gates each. **The seam guard existed but was never asserted here.** Every cycle arms a racing completion; nothing checked one fired. `PartitionStateCommitEncodeShift894Test` asserts it, this class did not. When #344 lands, the fixed encoder stops calling `getIncompleteOffsetsBelowHighestSucceeded()` - the method `RacingCommitCycleState` overrides to inject the race - and because the fix under test already produces zero shift and zero overshoot, all four tests here would have stayed green against a seam that never fires. `assertRaceFired` now runs after every commit that armed one, and names the re-hook in its failure message. `armRaceOn` clears the fired flag, so the guard is per-arm rather than per-instance: the repeating tests arm once per cycle, and a latched flag would report a seam that had since gone dead as healthy from the first cycle onward. **The manifest contradicted itself about who carries confluentinc#893.** `cherry-pick-892-pcmetrics-leak` still said PR #57 bundles it, while `cherry-pick-893-offset-reset` says #337 does. The manifest is the source of truth for fork-to-upstream ownership, so a collision check or release pass reading the first entry would have gone to the wrong PR. Corrected to name #337 as carrier and point at the owning entry; confluentinc#905 genuinely is still in #57 and stays.

…ey is visible Adds the shards.max.size gauge: the largest number of records queued in any single shard. shards.size already reported the total across all shards, which cannot distinguish an evenly loaded consumer from one where a single hot key is serialising all the work behind it - the case that actually hurts under KEY ordering. Cherry-pick of upstream confluentinc#905. Its test assertion travels in the un-quarantine commit rather than this one, because that commit rewrites the whole of PCMetricsTest and splitting the file between them would leave neither readable. Refs: confluentinc#905

…it on offsets The test was quarantined as flapping because it asserted PARTITION_LAST_COMMITTED_OFFSET against a shared completion COUNTER while the suite runs UNORDERED. Commits are contiguous and bounded by the lowest incomplete offset; completions are not ordered. Workers awaited a latch BEFORE incrementing, so a latched worker's offset never completed and the gap was permanent - the 120s atMost could not close it, only make the failure expensive. It passed only when the latched workers happened to hold the highest offsets, so a pass proved nothing. Rewritten to gate on each record's OWN offset rather than a shared counter, with two latched ranges held open per partition: a deliberate non-contiguous hole, and a freeze above a fixed offset. The pool is asserted wider than the workers the hole parks forever, because getting that wrong hangs rather than fails. Also carries the confluentinc#905 gauge assertion, since this commit rewrites the file wholesale and splitting it would leave neither half readable. Refs: #120, confluentinc#905

…ey is visible Adds the shards.max.size gauge: the largest number of records queued in any single shard. shards.size already reported the total across all shards, which cannot distinguish an evenly loaded consumer from one where a single hot key is serialising all the work behind it - the case that actually hurts under KEY ordering. Cherry-pick of upstream confluentinc#905. Its test assertion travels in the un-quarantine commit rather than this one, because that commit rewrites the whole of PCMetricsTest and splitting the file between them would leave neither readable. Refs: confluentinc#905

…it on offsets The test was quarantined as flapping because it asserted PARTITION_LAST_COMMITTED_OFFSET against a shared completion COUNTER while the suite runs UNORDERED. Commits are contiguous and bounded by the lowest incomplete offset; completions are not ordered. Workers awaited a latch BEFORE incrementing, so a latched worker's offset never completed and the gap was permanent - the 120s atMost could not close it, only make the failure expensive. It passed only when the latched workers happened to hold the highest offsets, so a pass proved nothing. Rewritten to gate on each record's OWN offset rather than a shared counter, with two latched ranges held open per partition: a deliberate non-contiguous hole, and a freeze above a fixed offset. The pool is asserted wider than the workers the hole parks forever, because getting that wrong hangs rather than fails. Also carries the confluentinc#905 gauge assertion, since this commit rewrites the file wholesale and splitting it would leave neither half readable. Refs: #120, confluentinc#905 The two live citations of the deleted note are repointed here rather than later, so no commit in this branch leaves a dangling reference behind - `bin/check-file-refs.sh` would fail any intermediate checkout otherwise, and a bisect landing there would blame the wrong change.