make committed offset accurate when partition assigned to avoid offset reset by sangreal · Pull Request #893 · confluentinc/parallel-consumer

Martyn Ye (sangreal) · 2025-10-20T08:31:14Z

Description...
make offsetHighestSucceeded accurate when partition assigned
We have encountered offset reset issue while frequent partition rebalancing.
The root cause is caused by :
(1) the offsetHighestSucceeded is assigned w/ offset in OffsetAndMetadata which is to-be processed
(2) incompletes is non-empty
(3) one WorkContainer is processed successfully, then dirty is true (this offset < offsetHighestSucceeded)
(4) committer choose (offsetHighestSucceeded + 1) to commit because incompletes is non-empty (the offset is removed from incompletes)
(5) rebalancing happens, new consumer try to pull record will throw out of range and begin offset reset

issue #894

Checklist

Documentation (if applicable)
Changelog

Roman Kolesnev (rkolesnev) · 2025-10-26T11:40:53Z

Hey Martyn Ye (@sangreal) - i will dig a bit more into it - but it doesn't look right to me - whenever partition is dirty (any offset was processed) - we do need to commit - even if highest succeeded was not advanced.
Something seems off to me in the steps above - there are two highest offsets that we care about - highest succeded overall and highest sequentially succeeded (that is offset that is committed) - when highest sequentially succeeded goes up - that is the base offset that is being committed. There cannot be any incompletes lower than that offset.

Roman Kolesnev (rkolesnev) · 2025-10-26T11:46:03Z

If you could build a test / reproducible example of this - it would help as well...

Martyn Ye (sangreal) · 2025-10-26T14:11:29Z

Roman Kolesnev (@rkolesnev) It is really great that you could take your personal time to review my pr 👍
Actually, initially I make the fix in here. https://github.com/confluentinc/parallel-consumer/blob/master/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/PartitionState.java#L207
to let this.offsetHighestSucceeded = this.offsetHighestSeen - 1;
But if rebalancing keeping happen and we have enough lower offset incompletes to commit, the offset will kept committing one by one and may eventually out of range.
So we have to make sure we only commit the processed message.
And when incompletes remove this offset before commit happens, offsetHighestSucceeded will be used instead.
So as you see, to make perfect fix, this will be very complex (make sure the offset removal happen after commit or other measure)

Again, I am all ear for your suggestion.

Roman Kolesnev (rkolesnev) · 2025-10-26T23:08:17Z

Hmm, I still don't fully understand where is the issue though.
In Kafka it is last processed offset + 1 that is committed - even in standard Kafka consumer. That should not cause issues. I don't think that we should be doing highe
Can you please review what you are observing and describe the flow step by step?
Is offset getting increased somewhere where there was no actual progress?

Martyn Ye (sangreal) · 2025-10-29T14:06:04Z

I have updated the way to fix after more thinking. The idea is make sure getOffsetToCommit is invoked only once to avoid dirty read and commit the wrong offset.
Roman Kolesnev (@rkolesnev)

Martyn Ye (sangreal) · 2025-10-30T03:33:19Z

John Byrne (@johnbyrnejb) please help review as well, thx a lot

Martyn Ye (sangreal) · 2025-10-31T09:13:41Z

Parallel Consumer Offset reset Issue flow.pdf
I have updated the diagram and pr again, please have a check when you have time, thanks a lot

Martyn Ye (sangreal) · 2025-11-05T03:28:27Z

Let me explain more for you guys to reviews.

supposed to commit 601266890 + 1 with incompletes (601266891)
after creating OffsetAndMetadata, incompletes (601266891) and complete offset with runlength encoding with result of : (11)
right before getOffsetToCommit, the 601266891 is processed and removed from incompletes. The offsetToCommit is changed to highestSucceedOffset + 1 since incompletes is empty (601266893)
As a result, the final offset (601266893) and OffsetMetadata (11)
rebalancing happened, when decoding OffsetAndMetadata, it will be calculated based on relative offset, therefore, highestSucceedOffset (601266894) and incompletes (601266893)
601266893 is normally processed, then incompletes be empty
when commit, the offsetToCommit is 601266894 + 1 = 601266895.
rebalancing happens again, when trying to poll 601266895, this offset doesn't exist, therefore offset reset happens.

Roman Kolesnev (@rkolesnev) John Byrne (@johnbyrnejb) please help review when you have time, we are waiting for the fix since this offset reset issue happens once every several days. Thanks a lot.

…tinc#893) Cherry-pick of confluentinc#893 (author: sangreal). Race condition in PartitionState: createOffsetAndMetadata() called tryToEncodeOffsets() and getOffsetToCommit() separately. Between the two calls, incompletes could drain, causing a higher offset to be committed than intended. After rebalance, the consumer fetches a non-existent offset and triggers auto.offset.reset (data loss or replay). Fix: tryToEncodeOffsets() now calls getOffsetToCommit() once at the top and returns a Tuple<Optional<String>, Long> so the offset and payload are computed atomically from the same state snapshot. Upstream PR: confluentinc#893 Approved by Roman Kolesnev, run in production for >1 week. Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Martyn Ye (sangreal) · 2026-05-11T09:44:07Z

John Byrne (@johnbyrnejb) Elisa Uhura (@elisauhura) could u please review the code since this is a valid bug when you have time, thx.

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>

…tinc#893) Cherry-pick of confluentinc#893 (author: sangreal). Race condition in PartitionState: createOffsetAndMetadata() called tryToEncodeOffsets() and getOffsetToCommit() separately. Between the two calls, incompletes could drain, causing a higher offset to be committed than intended. After rebalance, the consumer fetches a non-existent offset and triggers auto.offset.reset (data loss or replay). Fix: tryToEncodeOffsets() now calls getOffsetToCommit() once at the top and returns a Tuple<Optional<String>, Long> so the offset and payload are computed atomically from the same state snapshot. Upstream PR: confluentinc#893 Approved by Roman Kolesnev, run in production for >1 week. 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>

…tinc#893) Cherry-pick of confluentinc#893 (author: sangreal). Race condition in PartitionState: createOffsetAndMetadata() called tryToEncodeOffsets() and getOffsetToCommit() separately. Between the two calls, incompletes could drain, causing a higher offset to be committed than intended. After rebalance, the consumer fetches a non-existent offset and triggers auto.offset.reset (data loss or replay). Fix: tryToEncodeOffsets() now calls getOffsetToCommit() once at the top and returns a Tuple<Optional<String>, Long> so the offset and payload are computed atomically from the same state snapshot. Upstream PR: confluentinc#893 Approved by Roman Kolesnev, run in production for >1 week. Co-Authored-By: Claude Opus 4.6 (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>

Bring #57 current with master (release-notes #72, deps refresh #73/#74, unit-suite parallelisation #68, refactoring backlog #67, self-hosted CI, etc.). Only CHANGELOG.adoc conflicted: the 0.6.0.0 Fixes now lists master confluentinc#892 (per-commit OffsetMapCodecManager fix) alongside this PR confluentinc#859 (List->Set + assignment-path caching) and confluentinc#893 - complementary, kept all three. Regenerated README.adoc from the merged CHANGELOG via the asciidoc-template plugin. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

…#893) Spy PartitionState and verify getOffsetToCommit() is invoked exactly once during createOffsetAndMetadata(); a regression to the pre-confluentinc#893 two-call flow (which on a concurrent onSuccess could commit an offset inconsistent with the encoded incompletes payload) would make it fail. Also asserts the committed offset is the sequential-succeeded base. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

…tinc#893 names itself The changelog-issue gate that master gained in #104 flags two entries on this branch. One is a real omission: upstream PR confluentinc#893 says "issue confluentinc#894" in its own body, and confluentinc#894 is "Offset reset when frequent rebalancing" - precisely the problem the entry describes. It now cites it. upstream-map.yaml recorded issues: [] for this cherry-pick, so the map was wrong rather than merely thin. Corrected to [894], with a note saying what it used to say, and last_checked moved to today. The other flagged entry, the shards.max.size metric, genuinely has no issue: upstream confluentinc#905 opens "I want to add a metric..." and cites nothing. Its related: [71] is "Health-checks", far too general to claim as the request. That one takes the gate's changelog-ref opt-out in the PR body rather than an invented citation - which is the trade the gate's own comments call for. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QqHpNSXC39ANv9kG1ZvUzn

…he plan Phase 2 asks for a rehearsal on three throwaway issues before the bulk run. It was instead run on the six the fork has actually worked on - upstream #233, confluentinc#326, confluentinc#857, confluentinc#859, confluentinc#894, confluentinc#912 - mirrored as #117-#122. Real content tests the scheme harder than low-traffic issues chosen for being safe. Four things it changed, all of which would have bitten the bulk run: - AUTO-CLOSE REPLACES THE POST-MERGE STEP. "Fixes #NNN" in the fixing PR closes the mirror on merge. That deleted an artefact outright: pr57-post-merge.md existed only because closing and announcing had to be remembered by hand. - "ALREADY FIXED" NEEDS THREE STATES, NOT TWO. Only three of the six could honestly be closed. #233 is partial, confluentinc#857 is two-thirds fixed with the original deadlock still open, and confluentinc#326's landed fix only MIGHT be the same bug. A binary field would have produced three wrong closures, so the script must never infer "fixed" from a linked PR existing. - UPSTREAM PRs NEED COMMENTS TOO, though they are correctly not mirrored. confluentinc#893 and confluentinc#905 are carried in #57; their authors deserve to know their work ships, and readers deserve the pointer. - POSTING UPSTREAM IS GATED. Bodies are drafted and reviewed, but bulk posting into confluentinc/parallel-consumer was refused by tooling permissions. Worth knowing before Phase 2 step 4 assumes 78 comments can just be fired. Also records fork_issue in upstream-map.yaml as each mirror is created rather than as an end-of-migration sweep, which keeps the map honest in the interim. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QqHpNSXC39ANv9kG1ZvUzn

The house convention is that a bare #NN means THIS fork and upstream is written `upstream #NN`. Bare `confluentinc#859` in source was merely ambiguous before; now that the mirror #120 exists - and fork confluentinc#859 will eventually exist and mean something else - it is wrong. 27 refs qualified across the Java sources and docs/refactoring.md this PR already touches: #200, #233, confluentinc#857, confluentinc#859, confluentinc#893, confluentinc#905. Comments and one assertion message only; no logic. This is the "correct opportunistically as files are touched" rule from #114's Phase 4, not the bulk rewrite that plan rejects. Deliberately left alone, because a mechanical sweep gets them wrong: - CHANGELOG.adoc - its own header states that entries below 0.6.0.0 predate the fork and their #NN already refer to upstream, and the 0.6.0.0 entries use fully qualified links. Rewriting either would be wrong. - upstream-map.yaml - every number in it is upstream by construction, so prefixing all of them is noise, not clarity. - AGENTS.md - the only hit is a commit-subject convention shown BY EXAMPLE; rewriting the example would silently change the documented convention. Note what this does NOT do: renumber references to fork issue numbers. #114 rejects that, and three of its four reasons still stand - git history is immutable, a mechanical remap cannot tell an issue from the #999/#123 test offsets in Java source, and fork numbering will eventually collide with upstream's range. The fourth reason, staying cherry-pickable upstream, no longer applies. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QqHpNSXC39ANv9kG1ZvUzn

Antony Stubbs (astubbs) · 2026-08-04T23:43:39Z

I am carrying this PR in my fork, in astubbs#57, because it is approved here but has not merged. Credit and authorship are unchanged - it is cherry-picked as-is.

The issue it fixes (#894) is mirrored there as astubbs#121.

The Confluent-hosted version of this project is no longer maintained - its README says so. Parallel Consumer itself continues in my fork: https://github.com/astubbs/parallel-consumer

Nothing in this repository is closed or changed by this comment.

The six mirrored issues (upstream #233, confluentinc#326, confluentinc#857, confluentinc#859, confluentinc#894, confluentinc#912) and the two carried PRs (confluentinc#893, confluentinc#905) all have their backlink comment as of 2026-08-05. Comment URLs recorded in upstream-map.yaml under forwarded: for the entries the map covers, plus fork_issue for confluentinc#857 -> #119 and confluentinc#912 -> #122. upstream #233 and confluentinc#326 have no map entry; their mirrors are the record. Two findings promoted into the plan, because both change what the Phase 1 script must do: POSTING NEEDS A HUMAN. Agent tooling refused to bulk-post into another org's repo, so these went out through a reviewed script run by hand. Phase 2 step 4 should not assume 78 comments can be fired by automation. WRITE FOR THE READER, NOT THE TRACKER. Five review passes cut about a third of the drafted text. What went: PR base branches and commit-counts, "so the fork mirror stays open" (our bookkeeping, not the reporter's problem), a walkthrough of LinkedHashSet vs List, and hedging - "a maintained community fork", "one of its original authors" - which understates a fork that is ~99% one person's commits. Two rules fall out: link the PR rather than describe its state, and never write a sentence that goes stale in a week. Also: "this project is no longer maintained" reads, on an upstream issue, as "Parallel Consumer is dead". Name the Confluent-hosted version specifically and say the project continues. Every one of the eight now does. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01QqHpNSXC39ANv9kG1ZvUzn

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

make offsetHighestSucceeded accurate when partition assigned

64cd5f4

Martyn Ye (sangreal) requested a review from a team as a code owner October 20, 2025 08:31

Martyn Ye (sangreal) changed the title ~~make offsetHighestSucceeded accurate when partition assigned~~ make offsetHighestSucceeded accurate when partition assigned to avoid offset reset Oct 20, 2025

Martyn Ye (sangreal) added 8 commits October 20, 2025 17:01

update readme

8e66cb1

fix header

c458c98

fix issue

52ca70a

add guard to avoid wrong commit

e280a55

add more comment

727a140

fix UT

7c6c8ba

revert comment

ff5c7b2

Merge branch 'master' into fix/offset-reset

7a59bf9

Martyn Ye (sangreal) added 4 commits October 28, 2025 23:15

lastProcessedOffset to track the last incompletes offset

9a21085

clear up lastProcessedOffset after commit

7e73047

add clear func

551c9cf

add update logic check

40fba3e

Martyn Ye (sangreal) changed the title ~~make offsetHighestSucceeded accurate when partition assigned to avoid offset reset~~ make committed accurate when partition assigned to avoid offset reset Oct 29, 2025

Martyn Ye (sangreal) changed the title ~~make committed accurate when partition assigned to avoid offset reset~~ make committed offset accurate when partition assigned to avoid offset reset Oct 29, 2025

Martyn Ye (sangreal) added 3 commits October 30, 2025 10:53

add null check

bbed50f

update readme

6fe9d12

revet

6c7711e

Martyn Ye (sangreal) added 2 commits October 31, 2025 09:07

Merge branch 'master' into fix/offset-reset

6057e16

fix race condition when getOffsetToCommit

d700529

Merge branch 'master' into fix/offset-reset

62cb2cb

claude Bot mentioned this pull request Aug 4, 2026

docs: audit the in-flight ledger against reality, then split it so PRs stop colliding astubbs/parallel-consumer#112

Merged

5 tasks

Antony Stubbs (astubbs) mentioned this pull request Aug 4, 2026

confluentinc#894: Offset reset when frequent rebalancing astubbs/parallel-consumer#121

Open

Antony Stubbs (astubbs) mentioned this pull request Aug 4, 2026

Offset reset when frequent rebalancing #894

Open

Antony Stubbs (astubbs) mentioned this pull request Aug 5, 2026

confluentinc#777: Handling Partition Revocation in Parallel-Consumer Leading to Duplicate Event Processing astubbs/parallel-consumer#173

Open

Navigation Menu

Uh oh!

make committed offset accurate when partition assigned to avoid offset reset - #893

make committed offset accurate when partition assigned to avoid offset reset#893
Martyn Ye (sangreal) wants to merge 19 commits into
confluentinc:masterfrom
sangreal:fix/offset-reset

Martyn Ye (sangreal) commented Oct 20, 2025 •

edited

Loading

Uh oh!

Roman Kolesnev (rkolesnev) commented Oct 26, 2025 •

edited

Loading

Uh oh!

Roman Kolesnev (rkolesnev) commented Oct 26, 2025

Uh oh!

Martyn Ye (sangreal) commented Oct 26, 2025

Uh oh!

Roman Kolesnev (rkolesnev) commented Oct 26, 2025

Uh oh!

Martyn Ye (sangreal) commented Oct 29, 2025 •

edited

Loading

Uh oh!

Martyn Ye (sangreal) commented Oct 30, 2025

Uh oh!

Martyn Ye (sangreal) commented Oct 31, 2025

Uh oh!

Martyn Ye (sangreal) commented Nov 5, 2025 •

edited

Loading

Uh oh!

Martyn Ye (sangreal) commented May 11, 2026

Uh oh!

Antony Stubbs (astubbs) commented Aug 4, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Uh oh!

Conversation

Martyn Ye (sangreal) commented Oct 20, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Checklist

Uh oh!

Roman Kolesnev (rkolesnev) commented Oct 26, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Roman Kolesnev (rkolesnev) commented Oct 26, 2025

Uh oh!

Martyn Ye (sangreal) commented Oct 26, 2025

Uh oh!

Roman Kolesnev (rkolesnev) commented Oct 26, 2025

Uh oh!

Martyn Ye (sangreal) commented Oct 29, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Martyn Ye (sangreal) commented Oct 30, 2025

Uh oh!

Martyn Ye (sangreal) commented Oct 31, 2025

Uh oh!

Martyn Ye (sangreal) commented Nov 5, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

Martyn Ye (sangreal) commented May 11, 2026

Uh oh!

Antony Stubbs (astubbs) commented Aug 4, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Martyn Ye (sangreal) commented Oct 20, 2025 •

edited

Loading

Roman Kolesnev (rkolesnev) commented Oct 26, 2025 •

edited

Loading

Martyn Ye (sangreal) commented Oct 29, 2025 •

edited

Loading

Martyn Ye (sangreal) commented Nov 5, 2025 •

edited

Loading