github-actions[bot] commented on code in PR #41115:
URL: https://github.com/apache/beam/pull/41115#discussion_r4514955144
##########
runners/flink/src/test/java/org/apache/beam/runners/flink/PortableTimersExecutionTest.java:
##########
@@ -104,14 +104,14 @@ public static void tearDown() throws InterruptedException
{
flinkJobExecutor = null;
}
- @Test(timeout = 120_000)
+ @Test(timeout = 240_000)
public void testTimerExecution() throws Exception {
FlinkPipelineOptions options =
PipelineOptionsFactory.fromArgs("--experiments=beam_fn_api").as(FlinkPipelineOptions.class);
options.setRunner(CrashingRunner.class);
options.setFlinkMaster("[local]");
options.setStreaming(isStreaming);
- options.setParallelism(2);
+ options.setParallelism(1);
Review Comment:
**🟠High**: In #41037, the root cause of the `streaming: false` timeout was
that in Flink batch mode, upstream stages (`Impulse -> Input` and `PAssert`)
can finish and call `FlinkExecutableStageFunction.close()` before downstream
stages behind blocking shuffles/sorts (`Timers` and `PAssert`) call `open()`.
With `environmentCacheMillis=0` (the default),
`ReferenceCountingExecutableStageContextFactory` immediately tears down the
embedded `FnHarness` when the reference count momentarily drops to `0`, causing
downstream stages to hang for 3 minutes waiting for a client connection.
Additionally, lowering `parallelism` from `2` to `1` reduces test coverage
for multi-partition timer execution (note that `PortableStateExecutionTest`
kept `setParallelism(2)` in #41037). We should keep `setParallelism(2)` and set
`setEnvironmentCacheMillis(10_000)` on `PortablePipelineOptions` to match
#41037.
Also, as done in #41037 (lines 195–211 below):
1. Pass a unique job ID (e.g., `"id-" + UUID.randomUUID()`) to
`createJobInvocation` to avoid
`FlinkExecutableStageContextFactory.jobFactories` collisions across
parameterized runs.
2. Poll `!JobInvocation.isTerminated(jobInvocation.getState())` in a `try /
finally` block that calls `jobInvocation.cancel()` if not terminated, so a
failed or timed-out job does not spin until timeout or leak onto the
single-threaded `flinkJobExecutor`.
```suggestion
options.setParallelism(2);
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
