fix(flink): enable batch read it for flink source v2

[HuangZhenQiu]

Describe the issue this Pull Request addresses

Fix the bug in Flink Hudi source v2 in batch mode

Summary and Changelog

1. Fix the bug in Flink Hudi source v2 in batch mode
2. Update test cases to reflect the change

Impact

none

Risk Level

none

Documentation Update

none

Contributor's checklist

• Read through contributor's guide
• Enough context is provided in the sections above
• Adequate tests were added if applicable

[danny0405]

which is required to drive the END_OF_INPUT signal
// in SourceReaderBase for bounded (batch) reads.

The original logic seems more reasonabe? The END_OF_INPUT should be signaled after all the records in the split are handled(or iterate over)?

[HuangZhenQiu]

I agree. The better way is to have a fake batch reader, and return the an empty batch records with finished split id inside.

[danny0405]

cc @cshuo did you know the standard way to handle the bounded source end up for source v2?

[cshuo]

The cause of the bug is that currently the SplitFetcher only deletes splits when the fetch result of a split reader is a RecordsWithSplitIds with empty records and a set of finished splits (to be deleted).

Actually, there is a callback onSplitFinished in HoodieSourceReader, where we can delete the finished splits explicitly.

  @Override
  protected void onSplitFinished(Map<String, HoodieSourceSplit> finishedSplitIds) {
    requestSplit(new ArrayList<>(finishedSplitIds.keySet()));
    // delete the finished splits.
    splitFetcherManager.removeSplits(new ArrayList<>(finishedSplitIds.values()));
  }

Note: there are some other related changes:

• Bug: HoodieSourceSplit#splitId() is not idempotent. The id should not contain mutable fields consumed and fileOffset.
• Handle SplitsRemoval event in HoodieSourceSplitReader#handleSplitsChanges, maybe just a log.

[hudi-bot]

CI report:

• b21af66 UNKNOWN
• d835741 UNKNOWN
• b63cb4d Azure: FAILURE
• ad2b01e UNKNOWN

Bot commands

@hudi-bot supports the following commands:

• @hudi-bot run azure re-run the last Azure build

[codecov-commenter]

Codecov Report

❌ Patch coverage is 75.60976% with 10 lines in your changes missing coverage. Please review.
✅ Project coverage is 66.27%. Comparing base (967e456) to head (ad2b01e).

Files with missing lines	Patch %	Lines
...he/hudi/source/reader/HoodieSourceSplitReader.java	68.00%	5 Missing and 3 partials ⚠️
.../apache/hudi/source/reader/DefaultBatchReader.java	77.77%	2 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #18325      +/-   ##
============================================
- Coverage     68.46%   66.27%   -2.20%     
+ Complexity    27472    26733     -739     
============================================
  Files          2427     2428       +1     
  Lines        132655   132681      +26     
  Branches      15994    15994              
============================================
- Hits          90820    87932    -2888     
- Misses        34784    37811    +3027     
+ Partials       7051     6938     -113     

Flag	Coverage Δ
common-and-other-modules	37.33% <75.60%> (-7.10%)	⬇️
hadoop-mr-java-client	45.06% <ø> (-0.01%)	⬇️
spark-client-hadoop-common	48.20% <ø> (-0.01%)	⬇️
spark-java-tests	48.82% <ø> (-0.01%)	⬇️
spark-scala-tests	44.95% <ø> (+<0.01%)	⬆️
utilities	38.63% <ø> (-0.01%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
...va/org/apache/hudi/source/reader/BatchRecords.java	97.14% <100.00%> (ø)
...rg/apache/hudi/source/split/HoodieSourceSplit.java	100.00% <100.00%> (ø)
.../apache/hudi/source/reader/DefaultBatchReader.java	77.77% <77.77%> (ø)
...he/hudi/source/reader/HoodieSourceSplitReader.java	76.08% <68.00%> (-13.92%)	⬇️

... and 119 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.