[client] Fix tiering hang on first_row merge engine empty batches

[Kaixuan-Duan]

Purpose

Linked issue: close #2371

Brief change log

• fluss-client
  • ScanRecords: add a Map<TableBucket, Long> nextLogOffsets field with a new two-arg constructor (legacy single-arg constructor preserved for backwards compatibility); expose two new accessors:
    • polledBuckets() — union of buckets that produced records and buckets that only advanced their next fetch offset.
    • nextLogOffset(bucket) — exclusive upper bound of consumed offsets in this poll round, or null if the bucket was not polled.
  • LogFetchCollector#collectFetch: always record the advanced nextFetchOffset per polled bucket, even when the materialized record list is empty, and pack it into the new ScanRecords constructor.
• fluss-flink-common
  • TieringSplitReader#forLogRecords: iterate scanRecords.polledBuckets() instead of scanRecords.buckets(). Determine end-of-range by comparing the scanner-reported nextLogOffset against the bucket's stoppingOffset (with the legacy lastRecord.logOffset() >= stoppingOffset - 1 check kept as a fallback for callers that don't supply nextLogOffset). Tolerate splits that finish with no real record observed by falling back to UNKNOWN_BUCKET_TIMESTAMP when computing the finish timestamp.

Tests

./mvnw -pl fluss-client,fluss-flink/fluss-flink-common \
-Dtest='ScanRecordsTest,LogFetchCollectorTest,TieringSplitReaderTest#testTieringFirstRowMergeEngineFinishes' \
-DfailIfNoTests=false test

API and Format

Documentation

[Kaixuan-Duan]

@luoyuxia @zuston Hi, I have tried to resolve issue #2371. PTAL

[Copilot]

Pull request overview

This PR addresses a tiering hang for tables using the FIRST_ROW merge engine where the log offset can advance via “empty” WAL batches (recordCount=0) even when no ScanRecord is materialized, causing the tiering layer to repeatedly poll the same range (Issue #2371).

Changes:

• Extend fluss-client ScanRecords to carry per-bucket nextLogOffsets, and expose polledBuckets() + nextLogOffset(bucket) so callers can observe offset progress even when no records were produced.
• Update LogFetchCollector to always record nextFetchOffset for each polled bucket and construct ScanRecords with these offsets.
• Update Flink TieringSplitReader#forLogRecords to iterate polledBuckets() and determine end-of-range using nextLogOffset (with fallback to last-record checks), plus add a regression test reproducing Issue #2371.

Reviewed changes

Copilot reviewed 6 out of 6 changed files in this pull request and generated 2 comments.

Show a summary per file

File	Description
fluss-flink/fluss-flink-common/src/test/java/org/apache/fluss/flink/tiering/source/TieringSplitReaderTest.java	Adds regression test to ensure tiering completes under FIRST_ROW with duplicate keys/empty batches.
fluss-flink/fluss-flink-common/src/main/java/org/apache/fluss/flink/tiering/source/TieringSplitReader.java	Uses polledBuckets() and nextLogOffset to finish splits even when only empty batches occur.
fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/ScanRecordsTest.java	Adds unit tests for legacy constructor behavior and new polledBuckets()/nextLogOffset semantics.
fluss-client/src/test/java/org/apache/fluss/client/table/scanner/log/LogFetchCollectorTest.java	Verifies empty/filtered responses still expose offset advancement via polledBuckets()/nextLogOffset.
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/ScanRecords.java	Introduces nextLogOffsets, new constructor, polledBuckets(), and nextLogOffset(bucket).
fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogFetchCollector.java	Always records advanced nextFetchOffset per polled bucket and returns it in ScanRecords.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Kaixuan-Duan]

@luoyuxia Hi, I have addressed the feedback. PTAL

[luoyuxia]

@Kaixuan-Duan Thanks for the pr. Will take a look when i got some time

[luoyuxia]

@Kaixuan-Duan Thanks for the pr. Left minor comments. PTAL

[Kaixuan-Duan]

@luoyuxia Thank you for the review. Addressed, PTAL

[luoyuxia]

Thanks. WIll have a review later

[luoyuxia]

cc @beryllw

[luoyuxia]

Will have a review today. Hope we can merge it in next week.

[luoyuxia]

@Kaixuan-Duan Thanks for that. And sorry for delay review. Left some comments again.

[Kaixuan-Duan]

@luoyuxia Thank you for the review. Addressed, PTAL.

[luoyuxia]

@Kaixuan-Duan Thanks. LGTM. Only one minor comment.

[Kaixuan-Duan]

@luoyuxia Hi, I’ve resolved the merge conflicts and reverted the commit you pushed. Please take a look when you have a chance. Thanks!

[luoyuxia]

+1