Lake tiering never advances for buckets whose pending log contains only empty WAL batches

[beryllw]

Search before asking

• I searched in the issues and found nothing similar.

Fluss version

main (development)

Please describe the bug 🐞

For a primary key table with lake tiering (Paimon) enabled, one bucket's tiering progress gets stuck forever:

• timestampLag metric stays at a constant value (e.g. 2h) and never recovers;
• pendingRecordsLag metric keeps growing;
• the tiering job logs show the same tiny log split (e.g. startingOffset=17, stoppingOffset=18)being regenerated and "finished" in every tiering round, but the bucket's lake log end offset in
  Fluss never moves past 17.

How to reproduce

1. Create a primary key table with table.datalake.enabled = true and start the tiering service.
2. Write some data and wait until it is tiered to the lake.
3. Send a write request whose records are all filtered out on the server, e.g. a DELETE for a non-existent key, or a duplicate upsert on a first_row merge engine table. The server appends an empty WAL batch that occupies one log offset (required to keep the writer batch sequence, see KvTablet#putAsLeader).
4. Stop writing to this bucket.

Expected: the empty offset is tiered (there is nothing to write), the bucket's lake log end offset advances past it, and the lag metrics go back to zero.

Actual: the split covering the empty batch completes without producing a write result, its offset is silently dropped by the committer, and the split generator recreates the identical split in every following round — an infinite loop with permanently wrong lag metrics.

Root cause (one line): TieringCommitOperator#commitWriteResults filters write results by writeResult != null before building the logEndOffsets map, so buckets that finished a split without writing any data never get their log end offset committed back to Fluss.

Solution

• In TieringCommitOperator, collect log end offsets from all finished buckets (only skip buckets with unknown progress, i.e. skipped splits), and commit them to Fluss even when no bucket produced data: commit an empty lake snapshot carrying the bucket offsets so the tiering progress is persisted on a new snapshot (keeps the offsets-in-snapshot-properties invariant used by getMissingLakeSnapshot recovery).
• Paimon: call ignoreEmptyCommit(false) in PaimonLakeCommitter (Paimon skips empty commits by default; the commit.force-create-snapshot table option is only honored by Paimon's Flink sink, not by TableCommitImpl#commit). Iceberg empty appends already create snapshots.
• No snapshot storm: once the offsets catch up, no split is generated for the bucket anymore, so the empty snapshot is committed exactly once per catch-up event.

Are you willing to submit a PR?

• I'm willing to submit a PR!