[flink] Batch Log Table scan fix #3972 - #4031
Open
binary-signal wants to merge 1 commit into
Open
Conversation
…apache#3972 Signed-off-by: Evan <binary-signal@users.noreply.github.com>
binary-signal
changed the title
binary-signal
changed the title
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
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 registeredwhenbuckets finish while writers keep appending, and can hang when the table is
empty because
EARLIEST_OFFSET+stoppingOffset=0is not treated as an emptysplit.
Brief change log
FlinkSourceSplitReader.forLogRecords, unsubscribe finished bounded logsplits (
subscribedBuckets/stoppingOffsets+logScanner.unsubscribe)so later appends are not reported for a split that
SourceReaderBasehasalready unregistered.
stoppingOffset == 0(and non-negativestartingOffset >= stoppingOffset)as an empty log split so batch scans of empty buckets finish immediately
instead of subscribing with the
EARLIEST_OFFSETsentinel.Tests
FlinkSourceSplitReaderTest#testSubscribeEmptySplitWithEarliestStartingOffset— empty bucket with
startingOffset=EARLIEST_OFFSET,stoppingOffset=0finishes 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
SourceReaderBasefinished-split contract.