[flink] Batch Log Table scan fix #3972

[binary-signal]

Purpose

Linked issue: close #3972

Batch Flink SQL reads of Fluss log tables can fail with
IllegalStateException: Have records for a split that was not registered when
buckets finish while writers keep appending, and can hang when the table is
empty because EARLIEST_OFFSET + stoppingOffset=0 is not treated as an empty
split.

Brief change log

• In FlinkSourceSplitReader.forLogRecords, unsubscribe finished bounded log
  splits (subscribedBuckets / stoppingOffsets + logScanner.unsubscribe)
  so later appends are not reported for a split that SourceReaderBase has
  already unregistered.
• Treat stoppingOffset == 0 (and non-negative startingOffset >= stoppingOffset)
  as an empty log split so batch scans of empty buckets finish immediately
  instead of subscribing with the EARLIEST_OFFSET sentinel.

Tests

• FlinkSourceSplitReaderTest#testSubscribeEmptySplitWithEarliestStartingOffset
  — empty bucket with startingOffset=EARLIEST_OFFSET, stoppingOffset=0
  finishes on the first fetch.
• FlinkSourceSplitReaderTest#testFinishedLogSplitIsNotFetchedAgain
  — after a bounded log split finishes, further appends to the same bucket are
  not returned under that split id.

API and Format

No. Internal Flink source reader behavior only; no public API or storage format
change.

Documentation

No new user-facing feature. Behavior aligns batch log scans with Flink's
SourceReaderBase finished-split contract.

[leekeiabstraction]

@binary-signal Thank you for your PR, however this seems like a duplicate, can you check that this has been resolved by this merged PR? #4007

[binary-signal]

@leekeiabstraction it has been resolved,

i don't know if it's worth cherry picking from this PR the following

• testFinishedLogSplitIsNotFetchedAgain : [flink] Fix batch log scan with empty buckets #4007 has no reader level test asserting that a finished split is never re-fetched after subsequent appends.

• isEmptyLogSplit, if you want to improve readability.