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.

Martyn Ye (sangreal) · 2025-11-17T11:54:02Z

Roman Kolesnev (@rkolesnev) Thanks a lot for the efforts on the review my code.
John Byrne (@johnbyrnejb) could u please review my code as well, since Roman gave a pass on my code?
Thanks a lot in advance.

Martyn Ye (sangreal) · 2025-12-03T01:21:35Z

Edward Vaisman (@eddyv) could u please take a look when you have time, thanks a lot in advance.

…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 confluentinc#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>

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

Roman Kolesnev (rkolesnev) approved these changes Nov 17, 2025

View reviewed changes

Antony Stubbs (astubbs) mentioned this pull request Apr 18, 2026

fix: make committed offset accurate on partition assignment (cherry-pick Confluent #893) astubbs/parallel-consumer#42

Closed

3 tasks

Merge branch 'master' into fix/offset-reset

62cb2cb

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

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 Nov 17, 2025

Uh oh!

Martyn Ye (sangreal) commented Dec 3, 2025

Uh oh!

Martyn Ye (sangreal) commented May 11, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 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 Nov 17, 2025

Uh oh!

Martyn Ye (sangreal) commented Dec 3, 2025

Uh oh!

Martyn Ye (sangreal) commented May 11, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 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