This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 31fb173d9cb [FLINK-38536][tests] Fix data race in FinalizeOnMasterTest
and ExecutionGraphFinishTest (#28350)
31fb173d9cb is described below
commit 31fb173d9cb0234c67214153fea11841fd36a38a
Author: Martijn Visser <[email protected]>
AuthorDate: Mon Jun 29 22:07:04 2026 +0200
[FLINK-38536][tests] Fix data race in FinalizeOnMasterTest and
ExecutionGraphFinishTest (#28350)
* [FLINK-38536][tests] Fix data race in FinalizeOnMasterTest
forMainThread() runs the async deployment callbacks inline on the I/O
thread, racing
with the test thread that schedules the execution graph. Run all execution
graph
interactions on a dedicated single-threaded main-thread executor instead,
and drop
the temporary debug logging from PR #27168.
Generated-by: Claude Opus 4.8 (1M context)
* [FLINK-38536][tests] Fix same data race in ExecutionGraphFinishTest
testJobFinishes() shares the forMainThread() wiring and the same
deployment-callback
race fixed in FinalizeOnMasterTest. Apply the same fix.
Generated-by: Claude Opus 4.8 (1M context)
---
.../executiongraph/ExecutionGraphFinishTest.java | 103 ++++++++++++++++-----
.../executiongraph/FinalizeOnMasterTest.java | 82 +++++++++++-----
2 files changed, 138 insertions(+), 47 deletions(-)
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphFinishTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphFinishTest.java
index a096ebabbc7..148ee0d101e 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphFinishTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionGraphFinishTest.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.executiongraph;
import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
import
org.apache.flink.runtime.concurrent.ComponentMainThreadExecutorServiceAdapter;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobGraphTestUtils;
@@ -34,58 +35,110 @@ import org.junit.jupiter.api.extension.RegisterExtension;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledExecutorService;
+import java.util.function.Supplier;
import static org.assertj.core.api.Assertions.assertThat;
/** Tests the finish behaviour of the {@link ExecutionGraph}. */
class ExecutionGraphFinishTest {
+ /**
+ * Dedicated single-threaded main-thread executor; using forMainThread()
here would run the
+ * deployment callbacks on the I/O thread and race the test (FLINK-38536).
+ */
+ @RegisterExtension
+ static final TestExecutorExtension<ScheduledExecutorService>
JM_MAIN_THREAD_EXECUTOR_RESOURCE =
+ TestingUtils.jmMainThreadExecutorExtension();
+
@RegisterExtension
static final TestExecutorExtension<ScheduledExecutorService>
EXECUTOR_RESOURCE =
TestingUtils.defaultExecutorExtension();
@Test
void testJobFinishes() throws Exception {
-
JobGraph jobGraph =
JobGraphTestUtils.streamingJobGraph(
ExecutionGraphTestUtils.createJobVertex("Task1", 2,
NoOpInvokable.class),
ExecutionGraphTestUtils.createJobVertex("Task2", 2,
NoOpInvokable.class));
+ final ComponentMainThreadExecutor mainThreadExecutor =
+
ComponentMainThreadExecutorServiceAdapter.forSingleThreadExecutor(
+ JM_MAIN_THREAD_EXECUTOR_RESOURCE.getExecutor());
+
SchedulerBase scheduler =
new DefaultSchedulerBuilder(
- jobGraph,
-
ComponentMainThreadExecutorServiceAdapter.forMainThread(),
- EXECUTOR_RESOURCE.getExecutor())
+ jobGraph, mainThreadExecutor,
EXECUTOR_RESOURCE.getExecutor())
.build();
ExecutionGraph eg = scheduler.getExecutionGraph();
- scheduler.startScheduling();
- ExecutionGraphTestUtils.switchAllVerticesToRunning(eg);
-
- Iterator<ExecutionJobVertex> jobVertices =
eg.getVerticesTopologically().iterator();
-
- ExecutionJobVertex sender = jobVertices.next();
- ExecutionJobVertex receiver = jobVertices.next();
-
- List<ExecutionVertex> senderVertices =
Arrays.asList(sender.getTaskVertices());
- List<ExecutionVertex> receiverVertices =
Arrays.asList(receiver.getTaskVertices());
+ runInMainThread(mainThreadExecutor, scheduler::startScheduling);
+ runInMainThread(
+ mainThreadExecutor, () ->
ExecutionGraphTestUtils.switchAllVerticesToRunning(eg));
+
+ final ExecutionJobVertex[] orderedVertices =
+ supplyInMainThread(
+ mainThreadExecutor,
+ () -> {
+ Iterator<ExecutionJobVertex> jobVertices =
+ eg.getVerticesTopologically().iterator();
+ return new ExecutionJobVertex[] {
+ jobVertices.next(), jobVertices.next()
+ };
+ });
+ ExecutionJobVertex sender = orderedVertices[0];
+ ExecutionJobVertex receiver = orderedVertices[1];
+
+ List<ExecutionVertex> senderVertices =
+ supplyInMainThread(
+ mainThreadExecutor, () ->
Arrays.asList(sender.getTaskVertices()));
+ List<ExecutionVertex> receiverVertices =
+ supplyInMainThread(
+ mainThreadExecutor, () ->
Arrays.asList(receiver.getTaskVertices()));
// test getNumExecutionVertexFinished
- senderVertices.get(0).getCurrentExecutionAttempt().markFinished();
- assertThat(sender.getNumExecutionVertexFinished()).isOne();
- assertThat(eg.getState()).isEqualTo(JobStatus.RUNNING);
-
- senderVertices.get(1).getCurrentExecutionAttempt().markFinished();
- assertThat(sender.getNumExecutionVertexFinished()).isEqualTo(2);
- assertThat(eg.getState()).isEqualTo(JobStatus.RUNNING);
+ runInMainThread(
+ mainThreadExecutor,
+ () ->
senderVertices.get(0).getCurrentExecutionAttempt().markFinished());
+ assertThat(supplyInMainThread(mainThreadExecutor,
sender::getNumExecutionVertexFinished))
+ .isOne();
+ assertThat(supplyInMainThread(mainThreadExecutor, eg::getState))
+ .isEqualTo(JobStatus.RUNNING);
+
+ runInMainThread(
+ mainThreadExecutor,
+ () ->
senderVertices.get(1).getCurrentExecutionAttempt().markFinished());
+ assertThat(supplyInMainThread(mainThreadExecutor,
sender::getNumExecutionVertexFinished))
+ .isEqualTo(2);
+ assertThat(supplyInMainThread(mainThreadExecutor, eg::getState))
+ .isEqualTo(JobStatus.RUNNING);
// test job finishes
- receiverVertices.get(0).getCurrentExecutionAttempt().markFinished();
- receiverVertices.get(1).getCurrentExecutionAttempt().markFinished();
- assertThat(eg.getNumFinishedVertices()).isEqualTo(4);
- assertThat(eg.getState()).isEqualTo(JobStatus.FINISHED);
+ runInMainThread(
+ mainThreadExecutor,
+ () -> {
+
receiverVertices.get(0).getCurrentExecutionAttempt().markFinished();
+
receiverVertices.get(1).getCurrentExecutionAttempt().markFinished();
+ });
+ assertThat(eg.waitUntilTerminal()).isEqualTo(JobStatus.FINISHED);
+ assertThat(supplyInMainThread(mainThreadExecutor,
eg::getNumFinishedVertices)).isEqualTo(4);
+ assertThat(supplyInMainThread(mainThreadExecutor, eg::getState))
+ .isEqualTo(JobStatus.FINISHED);
+ }
+
+ /** Runs the action on the JobManager main thread and blocks until it
completes. */
+ private static void runInMainThread(
+ final ComponentMainThreadExecutor mainThreadExecutor, final
Runnable action)
+ throws Exception {
+ CompletableFuture.runAsync(action, mainThreadExecutor).get();
+ }
+
+ /** Reads the value on the JobManager main thread and blocks until it is
available. */
+ private static <T> T supplyInMainThread(
+ final ComponentMainThreadExecutor mainThreadExecutor, final
Supplier<T> supplier)
+ throws Exception {
+ return CompletableFuture.supplyAsync(supplier,
mainThreadExecutor).get();
}
}
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/FinalizeOnMasterTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/FinalizeOnMasterTest.java
index 9a2066fdfd0..a2b792ee239 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/FinalizeOnMasterTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/FinalizeOnMasterTest.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.executiongraph;
import org.apache.flink.api.common.JobStatus;
+import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
import
org.apache.flink.runtime.concurrent.ComponentMainThreadExecutorServiceAdapter;
import org.apache.flink.runtime.jobgraph.JobGraphTestUtils;
import org.apache.flink.runtime.jobgraph.JobVertex;
@@ -30,10 +31,10 @@ import
org.apache.flink.testutils.executor.TestExecutorExtension;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledExecutorService;
+import java.util.function.Supplier;
import static
org.apache.flink.runtime.scheduler.SchedulerTestingUtils.createScheduler;
import static org.assertj.core.api.Assertions.assertThat;
@@ -47,7 +48,14 @@ import static org.mockito.Mockito.verify;
* only when the execution graph reaches the successful final state.
*/
class FinalizeOnMasterTest {
- private static final Logger LOG =
LoggerFactory.getLogger(FinalizeOnMasterTest.class);
+
+ /**
+ * Dedicated single-threaded main-thread executor; using forMainThread()
here would run the
+ * deployment callbacks on the I/O thread and race the test (FLINK-38536).
+ */
+ @RegisterExtension
+ static final TestExecutorExtension<ScheduledExecutorService>
JM_MAIN_THREAD_EXECUTOR_RESOURCE =
+ TestingUtils.jmMainThreadExecutorExtension();
@RegisterExtension
static final TestExecutorExtension<ScheduledExecutorService>
EXECUTOR_RESOURCE =
@@ -63,31 +71,35 @@ class FinalizeOnMasterTest {
vertex2.setInvokableClass(NoOpInvokable.class);
vertex2.setParallelism(2);
+ final ComponentMainThreadExecutor mainThreadExecutor =
+
ComponentMainThreadExecutorServiceAdapter.forSingleThreadExecutor(
+ JM_MAIN_THREAD_EXECUTOR_RESOURCE.getExecutor());
+
final SchedulerBase scheduler =
createScheduler(
JobGraphTestUtils.streamingJobGraph(vertex1, vertex2),
-
ComponentMainThreadExecutorServiceAdapter.forMainThread(),
+ mainThreadExecutor,
EXECUTOR_RESOURCE.getExecutor());
- scheduler.startScheduling();
final ExecutionGraph eg = scheduler.getExecutionGraph();
- if (!eg.getState().equals(JobStatus.RUNNING)) {
- ErrorInfo ei = eg.getFailureInfo();
- LOG.info("Unexpected state found: Exception as string " +
ei.getExceptionAsString());
- LOG.info("Unexpected state found: ErrorInfo as string " + ei);
- }
- assertThat(eg.getState()).isEqualTo(JobStatus.RUNNING);
- ExecutionGraphTestUtils.switchAllVerticesToRunning(eg);
+ runInMainThread(mainThreadExecutor, scheduler::startScheduling);
+ assertThat(supplyInMainThread(mainThreadExecutor, eg::getState))
+ .isEqualTo(JobStatus.RUNNING);
+
+ runInMainThread(
+ mainThreadExecutor, () ->
ExecutionGraphTestUtils.switchAllVerticesToRunning(eg));
// move all vertices to finished state
- ExecutionGraphTestUtils.finishAllVertices(eg);
+ runInMainThread(mainThreadExecutor, () ->
ExecutionGraphTestUtils.finishAllVertices(eg));
assertThat(eg.waitUntilTerminal()).isEqualTo(JobStatus.FINISHED);
+ // waitUntilTerminal acts as the barrier, so the assertions below are
safe off the main
+ // thread.
verify(vertex1,
times(1)).finalizeOnMaster(any(FinalizeOnMasterContext.class));
verify(vertex2,
times(1)).finalizeOnMaster(any(FinalizeOnMasterContext.class));
- assertThat(eg.getRegisteredExecutions()).isEmpty();
+ assertThat(supplyInMainThread(mainThreadExecutor,
eg::getRegisteredExecutions)).isEmpty();
}
@Test
@@ -96,28 +108,54 @@ class FinalizeOnMasterTest {
vertex.setInvokableClass(NoOpInvokable.class);
vertex.setParallelism(1);
+ final ComponentMainThreadExecutor mainThreadExecutor =
+
ComponentMainThreadExecutorServiceAdapter.forSingleThreadExecutor(
+ JM_MAIN_THREAD_EXECUTOR_RESOURCE.getExecutor());
+
final SchedulerBase scheduler =
createScheduler(
JobGraphTestUtils.streamingJobGraph(vertex),
-
ComponentMainThreadExecutorServiceAdapter.forMainThread(),
+ mainThreadExecutor,
EXECUTOR_RESOURCE.getExecutor());
- scheduler.startScheduling();
final ExecutionGraph eg = scheduler.getExecutionGraph();
- assertThat(eg.getState()).isEqualTo(JobStatus.RUNNING);
+ runInMainThread(mainThreadExecutor, scheduler::startScheduling);
+ assertThat(supplyInMainThread(mainThreadExecutor, eg::getState))
+ .isEqualTo(JobStatus.RUNNING);
- ExecutionGraphTestUtils.switchAllVerticesToRunning(eg);
+ runInMainThread(
+ mainThreadExecutor, () ->
ExecutionGraphTestUtils.switchAllVerticesToRunning(eg));
// fail the execution
- final Execution exec =
-
eg.getJobVertex(vertex.getID()).getTaskVertices()[0].getCurrentExecutionAttempt();
- exec.fail(new Exception("test"));
+ runInMainThread(
+ mainThreadExecutor,
+ () -> {
+ final Execution exec =
+ eg.getJobVertex(vertex.getID())
+ .getTaskVertices()[0]
+ .getCurrentExecutionAttempt();
+ exec.fail(new Exception("test"));
+ });
assertThat(eg.waitUntilTerminal()).isEqualTo(JobStatus.FAILED);
verify(vertex,
times(0)).finalizeOnMaster(any(FinalizeOnMasterContext.class));
- assertThat(eg.getRegisteredExecutions()).isEmpty();
+ assertThat(supplyInMainThread(mainThreadExecutor,
eg::getRegisteredExecutions)).isEmpty();
+ }
+
+ /** Runs the action on the JobManager main thread and blocks until it
completes. */
+ private static void runInMainThread(
+ final ComponentMainThreadExecutor mainThreadExecutor, final
Runnable action)
+ throws Exception {
+ CompletableFuture.runAsync(action, mainThreadExecutor).get();
+ }
+
+ /** Reads the value on the JobManager main thread and blocks until it is
available. */
+ private static <T> T supplyInMainThread(
+ final ComponentMainThreadExecutor mainThreadExecutor, final
Supplier<T> supplier)
+ throws Exception {
+ return CompletableFuture.supplyAsync(supplier,
mainThreadExecutor).get();
}
}
