[flink] Fix batch log scan with empty buckets

[Gezi-lzq]

Purpose

Linked issue: close #3972

When a Flink batch job scans a multi-bucket Fluss Log Table with low parallelism, an empty bucket can remain active after its bounded split has finished. Subsequent fetch results may then contain a split ID that Flink has already unregistered.

This change ensures that bounded log splits are completed based on scanner progress and that finished buckets are retired before the next fetch.

Brief change log

• Complete bounded log splits when consumedUpToOffset reaches the stopping offset.
• Handle empty bounded splits starting from EARLIEST_OFFSET.
• Remove finished buckets from the Flink source reader mapping and unsubscribe them from the log scanner.
• Add regression coverage for empty buckets and empty bounded splits.

Tests

• FlinkSourceSplitReaderTest
• Flink120TableSourceBatchITCase
• ./mvnw -pl fluss-flink/fluss-flink-common -DskipTests verify

API and Format

No API or storage format changes.

Documentation

No documentation changes.

Generative AI Disclosure

• Yes — OpenAI Codex

[Gezi-lzq]

Hi @loserwang1024 and @wuchong, when you have time, could you please help review this PR?

[loserwang1024]

I have left some tiny comment

[loserwang1024]

In which situation, stoppingOffset == 0?

[Gezi-lzq]

This covers a newly created table or partition with no data, where the latest offset is 0. Since EARLIEST_OFFSET is the sentinel value -2, startingOffset >= stoppingOffset evaluates to false even though there are no records to read. Therefore, we need to handle stoppingOffset == 0 explicitly.

[loserwang1024]

Maybe we can unsubscribe each finished scanBucket directly in forLogRecords here, avoiding duplicate remove logic in two places(removeFinishedSplits).

[Gezi-lzq]

Thanks for the suggestion. I kept the unsubscribe logic in fetch() because forLogRecords() is currently responsible for converting ScanRecords into FlinkRecordsWithSplitIds. Moving it there would introduce scanner-state side effects and mix result construction with lifecycle cleanup. There is no duplicate removal implementation here; removeFinishedSplits() remains the single cleanup helper.

[Gezi-lzq]

Thanks @loserwang1024 for taking the time to review this PR.

This PR incorporates the fix from PR #3873 for issue #3872, using consumedUpToOffset >= stoppingOffset to complete bounded splits when filtering produces no materialized records but scanner progress has reached the end, preventing batch jobs from hanging. It further fixes issue #3972, where an empty bucket in a multi-bucket Log Table could cause batch scans to fail under low parallelism.