damccorm opened a new pull request, #41037: URL: https://github.com/apache/beam/pull/41037
### Summary Fixes flaky timeouts and cascading session-leak failures in `PreCommit Java` (e.g. [run 42235327251](https://github.com/apache/beam/actions/runs/42235327251)) across: - `:runners:spark:3:test` (`StructuredStreamingPipelineStateTest.testBatchCancelStopsRunningJob` and `testCancelFollowedImmediatelyBySecondPipeline`) - `:runners:flink:1.19:test` (`PortableStateExecutionTest.testExecution`) ### Verification Workflow Run - Triggered `PreCommit Java` on `fix/precommit-java-spark-flink-state-test-flakes`: https://github.com/apache/beam/actions/runs/42244720039 ### Root Cause 1. In `StructuredStreamingPipelineStateTest`, `runBlockingPipeline()` waited 60s for `BlockingDoFn` to start and threw `AssertionError("DoFn did not start")` without calling `result.cancel()` / `result.waitUntilFinish()`. Under heavy CPU contention in `PreCommit Java`, cold `SparkSession` initialization can exceed 60s, and when `runBlockingPipeline()` threw without cancelling `result`, the background `BlockingDoFn` and owned `SparkSession` leaked into subsequent tests in the same JVM (`testCancelFollowedImmediatelyBySecondPipeline`), causing `AssertionError: session not stopped`. 2. In `PortableStateExecutionTest`, `testExecution` polled `while (jobInvocation.getState() != JobState.Enum.DONE)` with a 120s JUnit timeout and did not check `JobInvocation.isTerminated(...)` or cancel the job on timeout/failure, leaving the single-threaded `flinkJobExecutor` blocked. ### Changes - In `StructuredStreamingPipelineStateTest.java`, increase `DEADLINE_SECONDS` from `60` to `180`, ensure `runBlockingPipeline()` always cancels and waits for `result` if `started.await(...)` does not succeed, and add an `@After` teardown to stop any leftover active/default `SparkSession`. - In `PortableStateExecutionTest.java`, increase `@Test(timeout = 300_000)`, poll `!JobInvocation.isTerminated(jobInvocation.getState())`, assert `JobState.Enum.DONE`, and cancel the job in `finally` if not terminated. -- 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]
