[BEAM-11021] Add unsupported BundleFinalizationHandler to portable batch Flink.

[boyuanzz]

This PR added an unsupported BundleFinalizationHandler to portable Flink in batch mode. This will give users a more detailed error message like:

ERROR:root:java.lang.UnsupportedOperationException: Portable Flink runner doesn't support bundle finalization in batch mode. For more details, please refer to https://issues.apache.org/jira/browse/BEAM-11021.

when users try to use bundle finalization in batch.

r: @mxm
cc: @lukecwik @ibzib

---

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

• Choose reviewer(s) and mention them in a comment (R: @username).
• Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
• 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.

Post-Commit Tests Status (on master branch)

Lang	SDK	Dataflow	Flink	Samza	Spark	Twister2
Go		---		---		---
Java
Python				---		---
XLang		---		---		---

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website	Whitespace	Typescript
Non-portable
Portable	---		---	---	---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

[codecov]

Codecov Report

Merging #13054 into master will decrease coverage by 0.00%.
The diff coverage is n/a.

@@            Coverage Diff             @@
##           master   #13054      +/-   ##
==========================================
- Coverage   82.49%   82.48%   -0.01%     
==========================================
  Files         455      455              
  Lines       54876    54876              
==========================================
- Hits        45270    45266       -4     
- Misses       9606     9610       +4     

Impacted Files	Coverage Δ
.../python/apache_beam/transforms/periodicsequence.py	96.49% <0.00%> (-1.76%)	⬇️
sdks/python/apache_beam/runners/common.py	88.75% <0.00%> (-0.45%)	⬇️

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 571338b...6b00b0b. Read the comment docs.

[boyuanzz]

Run Java PreCommit

[mxm]

Test failure, but looks good otherwise.

[boyuanzz]

Test failure, but looks good otherwise.

Thanks for the quick approval! The spark unit tests were complaining mocking ambiguous stageBundleFactory.getBundle() method.

[boyuanzz]

Run Python_PVR_Flink PreCommit

[boyuanzz]

Run Python_PVR_Flink PreCommit

[boyuanzz]

Run Python_PVR_Flink PreCommit

[boyuanzz]

Flink test failures are cased by #13014

[boyuanzz]

Run Python_PVR_Flink PreCommit

[mxm]

Test Result (1 failure / +1)

    apache_beam.runners.portability.flink_runner_test.FlinkRunnerTestStreaming.test_register_finalizations

https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Phrase/16/

This is not expected?

[boyuanzz]

Run Python_PVR_Flink PreCommit

[boyuanzz]

Test Result (1 failure / +1)

    apache_beam.runners.portability.flink_runner_test.FlinkRunnerTestStreaming.test_register_finalizations

https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Phrase/16/

This is not expected?

This is not expected. It seems like the test is kind of flaky .

[boyuanzz]

Thanks for your quick review! I'm going to merge this PR.