[BEAM-10120] Add dynamic timer support to portable Flink

[boyuanzz]

Please add a meaningful description for your change here

---

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.

[boyuanzz]

Run Java Flink PortableValidatesRunner Streaming

[boyuanzz]

Run Java Flink PortableValidatesRunner Batch

[boyuanzz]

Run Java Spark PortableValidatesRunner Batch

[boyuanzz]

Run Java Spark PortableValidatesRunner Streaming

[boyuanzz]

r: @ibzib @mxm
cc: @y1chi @tysonjh

[codecov]

Codecov Report

Merging #13783 (b8c8bf3) into master (a724602) will decrease coverage by 0.01%.
The diff coverage is n/a.

@@            Coverage Diff             @@
##           master   #13783      +/-   ##
==========================================
- Coverage   82.76%   82.74%   -0.02%     
==========================================
  Files         466      466              
  Lines       57525    57525              
==========================================
- Hits        47612    47601      -11     
- Misses       9913     9924      +11     

Impacted Files	Coverage Δ
sdks/python/apache_beam/utils/interactive_utils.py	88.09% <0.00%> (-4.77%)	⬇️
...hon/apache_beam/runners/direct/test_stream_impl.py	91.91% <0.00%> (-2.21%)	⬇️
sdks/python/apache_beam/internal/metrics/metric.py	86.45% <0.00%> (-1.05%)	⬇️
sdks/python/apache_beam/io/localfilesystem.py	91.66% <0.00%> (-0.76%)	⬇️
...eam/runners/interactive/interactive_environment.py	89.92% <0.00%> (-0.36%)	⬇️
...ks/python/apache_beam/runners/worker/sdk_worker.py	89.85% <0.00%> (-0.32%)	⬇️
sdks/python/apache_beam/transforms/util.py	95.66% <0.00%> (-0.18%)	⬇️

---

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 a724602...b8c8bf3. Read the comment docs.

[boyuanzz]

Run Python_PVR_Flink PreCommit

[boyuanzz]

It seems like Spark doesn't support timers in portable execution. The changes I have made to java runner shared library can also used by Spark in the future. @ibzib

[ibzib]

It seems like Spark doesn't support timers in portable execution. The changes I have made to java runner shared library can also used by Spark in the future. @ibzib

Portable Spark supports timers in batch, but not streaming mode.

[boyuanzz]

Run Java Flink PortableValidatesRunner Streaming

[boyuanzz]

Run Java Flink PortableValidatesRunner Batch

[boyuanzz]

Run Java Flink PortableValidatesRunner Streaming

[boyuanzz]

Run Python_PVR_Flink PreCommit

[ibzib]

LGTM!

[boyuanzz]

Thanks for the review!

I'm going to merge this PR. Feel free to drop any comments here if you have any concerns @mxm . I'll follow up with any comments.