[flink] Preserve tiering offsets for empty-batch buckets
- Dominant language
- Java
- Stars
- 2.1k
- Forks
- 625
- Avg merge
- 3d 14h
- Merged PRs (30d)
- 97
Description
### Search before asking
- [x] I searched in the [issues](https://github.com/apache/fluss/issues) and found nothing similar.
### Fluss version
main (development)
### Please describe the bug 🐞
Related to #2371 and the fix being discussed in #3242.
When tiering a table where some buckets produce lake write results and other buckets only advance through empty materialized batches, the empty-batch buckets can lose their advanced log offsets during commit.
Triggering sequence:
1. A tiering round finishes multiple bucket splits for the same table.
2. At least one bucket writes real records to the lake and returns a non-null `TableBucketWriteResult.writeResult()`.
3. At least one other bucket reaches its stopping offset through empty batches, for example duplicate upserts under the `FIRST_ROW` merge engine, and returns a `TableBucketWriteResult` with a valid `logEndOffset` but `writeResult() == null`.
4. `TieringCommitOperator` filters the completed bucket results to non-empty write results before collecting `logEndOffsets`.
5. The lake commit proceeds because some buckets are non-empty, but the offset-only bucket is absent from the `logEndOffsets` committed to Fluss.
Expected behavior: when a commit contains any lake data, Fluss should commit log offset metadata for all completed buckets that have a known `logEndOffset`, including buckets that only advanced through empty batches.
Actual behavior: offset-only buckets are filtered out together with their null lake write result, so their committed lake offset may not advance even though the split finished.
### Solution
Keep using only non-null lake write results for the lake committer, but collect `logEndOffsets` and max tiered timestamps from all completed `TableBucketWriteResult`s whose `logEndOffset` is known.
This intentionally does not add support for a fully offset-only table commit where every bucket has `writeResult() == null`; it only fixes the mixed case where the same commit already contains lake data from other buckets.
### Are you willing to submit a PR?
- [x] I'm willing to submit a PR!
Contributor guide
No contributing guide indexed for this repository
Research direction
Start in TieringCommitOperator and trace how completed TableBucketWriteResults are filtered before lake commits. Preserve non-null write results for the lake committer while collecting known logEndOffset values and maximum tiered timestamps from all completed results; done when the mixed non-empty and empty-batch case commits offsets for every completed bucket.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- data-engineering, stream-processing
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 72/100