Flink testParDoRequiresStableInput flaky

[damccorm]

https://ci-beam.apache.org/job/beam_PreCommit_Java_Commit/20253/

org.apache.beam.runners.flink.FlinkRequiresStableInputTest.testParDoRequiresStableInput

java.util.concurrent.ExecutionException: java.util.concurrent.CompletionException: org.apache.flink.runtime.checkpoint.CheckpointException: Checkpoint triggering task Source: Impulse -> CreatePCollectionOfOneValue/Read(CreateSource)/ParDo(OutputSingleSource)/ParMultiDo(OutputSingleSource) -> CreatePCollectionOfOneValue/Read(CreateSource)/ParDo(BoundedSourceAsSDFWrapper)/ParMultiDo(BoundedSourceAsSDFWrapper)/Pair with initial restriction/ParMultiDo(PairWithRestriction) -> CreatePCollectionOfOneValue/Read(CreateSource)/ParDo(BoundedSourceAsSDFWrapper)/ParMultiDo(BoundedSourceAsSDFWrapper)/Split restriction/ParMultiDo(SplitRestriction) -> CreatePCollectionOfOneValue/Read(CreateSource)/ParDo(BoundedSourceAsSDFWrapper)/ParMultiDo(BoundedSourceAsSDFWrapper)/Explode windows/ParMultiDo(ExplodeWindows) -> CreatePCollectionOfOneValue/Read(CreateSource)/ParDo(BoundedSourceAsSDFWrapper)/ParMultiDo(BoundedSourceAsSDFWrapper)/Assign unique key/AddKeys/Map/ParMultiDo(Anonymous) -> ToKeyedWorkItem (1/1) of job 7bbb425ba325dbc1dc4d3cdf1c8b88f9 is not being executed at the moment. Aborting checkpoint. Failure reason: Not all required tasks are currently running.

Imported from Jira BEAM-13575. Original Jira may contain additional context.
Reported by: ibzib.

[kennknowles]

@je-ik is working on support for this in some cases. Might know something about this one.

[je-ik]

This does not seem related to the functionality itself, looks like the test is missing some synchronization. I'll take a look.

[Abacn]

It's appearing again in https://ci-beam.apache.org/job/beam_PreCommit_Java_Phrase/6030/

Test Result (2 failures / +2)
org.apache.beam.runners.flink.FlinkRequiresStableInputTest.testParDoRequiresStableInputPortable
org.apache.beam.runners.flink.FlinkRequiresStableInputTest.testParDoRequiresStableInputStatefulPortable

Error message:

org.junit.runners.model.TestTimedOutException: test timed out after 30000 milliseconds
	at java.lang.Thread.sleep(Native Method)
	at org.apache.beam.runners.flink.FlinkRequiresStableInputTest.executePipeline(FlinkRequiresStableInputTest.java:187)
	at org.apache.beam.runners.flink.FlinkRequiresStableInputTest.runTest(FlinkRequiresStableInputTest.java:160)
	at org.apache.beam.runners.flink.FlinkRequiresStableInputTest.testParDoRequiresStableInputPortable(FlinkRequiresStableInputTest.java:112)

[Abacn]

Downgrade to P2 as the issue is more likely due to Jenkins under high load. The functionality itself is implemented in #22889