apache / apache/fluss

[flink] Preserve tiering offsets for empty-batch buckets

Open
#3,413 0 comments 0 reactions 0 assignees View on GitHub
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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.