Enable drain operation for flink jobs with requiresStableInput annotation

[kkdoon]

Currently, drain operation does not work for flink pipelines when RequiresStableInput annotation is used. This is caused due to buffered data not being processed before the final checkpoint operation that causes watermark hold related exception. This PR addresses this issue by processing the buffer before the final checkpoint completes. More context in #28554.

---

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

• Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
• Update CHANGES.md with noteworthy changes.
• If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

[kkdoon]

Run PreCommit Java PVR Flink Batch

[kkdoon]

R: @mxm @je-ik

Could you kindly help me review this PR. Thanks!

[github-actions]

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control

[je-ik]

I don't think we can just skip the check. The fact that the check fails implies there are still buffered data that have not been processed yet. We need to either:
a) correctly flush and process the buffered data (which will clear the hold), or
b) ensure that checkpoint is called before the final watermark

I have a suspicion that correct is only b), because otherwise, in case of failure and restore from checkpoint after we flushed the data, we might break the stable input contract.

[kkdoon]

I don't think we can just skip the check. The fact that the check fails implies there are still buffered data that have not been processed yet. We need to either: a) correctly flush and process the buffered data (which will clear the hold), or b) ensure that checkpoint is called before the final watermark

I have a suspicion that correct is only b), because otherwise, in case of failure and restore from checkpoint after we flushed the data, we might break the stable input contract.

I agree that option B is the cleanest and makes logical sense. I think in implementation terms it works if somehow flushData is invoked after last checkpoint finishes (or some other approach). Option A violates the requiresStableInput contract and hence seems incorrect (i did experiment with this approach as well and it gave the expected output when checkpoint successfully completed).

Having said that, the current approach also works correctly when requiresStableInput is set since all the buffered data is emitted after checkpoint completes and the output watermark finally becomes equal to MAX_WATERMARK (i have run multiple jobs to verify the behavior). I can defer the currentOutputWatermark < Long.MAX_VALUE check for requiresStableInput scenario and do this check inside notifyCheckpointComplete, at the time of job completion, to verify that output watermark is correct.

[je-ik]

Having said that, the current approach also works correctly when requiresStableInput is set since all the buffered data is emitted after checkpoint completes and the output watermark finally becomes equal to MAX_WATERMARK (i have run multiple jobs to verify the behavior).

This seems to be the source of troubles. See my comment here.

[kkdoon]

@je-ik i have updated the PR based on our discussion. Could you kindly check again?

[je-ik]

LGTM, thanks! Can you please update the title of the PR to match the implementation? We can merge it afterwards.

[je-ik]

Could you also please squash the commits? Thanks!

[codecov]

Codecov Report

Merging #28567 (e08578a) into master (9fdc59b) will decrease coverage by 8.41%.
Report is 40 commits behind head on master.
The diff coverage is n/a.

@@            Coverage Diff             @@
##           master   #28567      +/-   ##
==========================================
- Coverage   38.39%   29.98%   -8.41%     
==========================================
  Files         686      391     -295     
  Lines      101640    65297   -36343     
==========================================
- Hits        39021    19580   -19441     
+ Misses      61040    45717   -15323     
+ Partials     1579        0    -1579     

Flag	Coverage Δ
go	?
python	29.98% <ø> (-0.03%)	⬇️

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

see 302 files with indirect coverage changes

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

[kkdoon]

closing this PR since i botched up squashing the commits and created #29102 instead