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 cbd7792ee9b [FLINK-38536][tests] Deploy synchronously to avoid
deployment-callback race
cbd7792ee9b is described below
commit cbd7792ee9b3c0e44124634d80f8aa1b55ba9734
Author: Martijn Visser <[email protected]>
AuthorDate: Tue Jun 30 10:38:19 2026 +0200
[FLINK-38536][tests] Deploy synchronously to avoid deployment-callback race
The FLINK-38536 rewrite (#28350) ran ExecutionGraphFinishTest and
FinalizeOnMasterTest on a dedicated single-thread main executor while keeping a
real future executor. Execution.deploy() offloads TaskDeploymentDescriptor
creation to that future executor and syncs the deploy-completion callback back
to the main thread asynchronously; tryGetTaskDeploymentDescriptorForSlot fails
the deployment when the execution is no longer DEPLOYING. The tests moved
executions to RUNNING via switchAllV [...]
Run scheduling and deployment synchronously on the test thread via
forMainThread() and DirectScheduledExecutorService so the deployment callbacks
complete inline while the executions are still DEPLOYING. This supersedes the
forSingleThreadExecutor approach from #28350 and removes the
runInMainThread/supplyInMainThread helpers it introduced.
Generated-by: Claude Opus 4.8 (1M context)
---
.../executiongraph/ExecutionGraphFinishTest.java | 113 +++++----------------
.../executiongraph/FinalizeOnMasterTest.java | 92 ++++-------------
2 files changed, 48 insertions(+), 157 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 148ee0d101e..1f8b40ee9ee 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,43 +19,25 @@
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;
import org.apache.flink.runtime.scheduler.DefaultSchedulerBuilder;
import org.apache.flink.runtime.scheduler.SchedulerBase;
import org.apache.flink.runtime.testtasks.NoOpInvokable;
-import org.apache.flink.testutils.TestingUtils;
-import org.apache.flink.testutils.executor.TestExecutorExtension;
+import org.apache.flink.runtime.testutils.DirectScheduledExecutorService;
import org.junit.jupiter.api.Test;
-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 =
@@ -63,82 +45,41 @@ class ExecutionGraphFinishTest {
ExecutionGraphTestUtils.createJobVertex("Task1", 2,
NoOpInvokable.class),
ExecutionGraphTestUtils.createJobVertex("Task2", 2,
NoOpInvokable.class));
- final ComponentMainThreadExecutor mainThreadExecutor =
-
ComponentMainThreadExecutorServiceAdapter.forSingleThreadExecutor(
- JM_MAIN_THREAD_EXECUTOR_RESOURCE.getExecutor());
-
+ // Use a direct future executor so the asynchronous deployment
callbacks complete inline
+ // before the manual state transitions below, avoiding the FLINK-38536
race.
SchedulerBase scheduler =
new DefaultSchedulerBuilder(
- jobGraph, mainThreadExecutor,
EXECUTOR_RESOURCE.getExecutor())
+ jobGraph,
+
ComponentMainThreadExecutorServiceAdapter.forMainThread(),
+ new DirectScheduledExecutorService())
.build();
ExecutionGraph eg = scheduler.getExecutionGraph();
- 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()));
+ scheduler.startScheduling();
+ ExecutionGraphTestUtils.switchAllVerticesToRunning(eg);
- // test getNumExecutionVertexFinished
- 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);
+ Iterator<ExecutionJobVertex> jobVertices =
eg.getVerticesTopologically().iterator();
- // test job finishes
- 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);
- }
+ ExecutionJobVertex sender = jobVertices.next();
+ ExecutionJobVertex receiver = jobVertices.next();
- /** 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();
- }
+ List<ExecutionVertex> senderVertices =
Arrays.asList(sender.getTaskVertices());
+ List<ExecutionVertex> receiverVertices =
Arrays.asList(receiver.getTaskVertices());
+
+ // test getNumExecutionVertexFinished
+ senderVertices.get(0).getCurrentExecutionAttempt().markFinished();
+ assertThat(sender.getNumExecutionVertexFinished()).isOne();
+ assertThat(eg.getState()).isEqualTo(JobStatus.RUNNING);
- /** 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();
+ senderVertices.get(1).getCurrentExecutionAttempt().markFinished();
+ assertThat(sender.getNumExecutionVertexFinished()).isEqualTo(2);
+ assertThat(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);
}
}
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 a2b792ee239..9e510ddb660 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,22 +19,15 @@
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;
import org.apache.flink.runtime.jobgraph.JobVertex.FinalizeOnMasterContext;
import org.apache.flink.runtime.scheduler.SchedulerBase;
import org.apache.flink.runtime.testtasks.NoOpInvokable;
-import org.apache.flink.testutils.TestingUtils;
-import org.apache.flink.testutils.executor.TestExecutorExtension;
+import org.apache.flink.runtime.testutils.DirectScheduledExecutorService;
import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.RegisterExtension;
-
-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;
@@ -49,18 +42,6 @@ import static org.mockito.Mockito.verify;
*/
class FinalizeOnMasterTest {
- /**
- * 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 testFinalizeIsCalledUponSuccess() throws Exception {
final JobVertex vertex1 = spy(new JobVertex("test vertex 1"));
@@ -71,35 +52,28 @@ class FinalizeOnMasterTest {
vertex2.setInvokableClass(NoOpInvokable.class);
vertex2.setParallelism(2);
- final ComponentMainThreadExecutor mainThreadExecutor =
-
ComponentMainThreadExecutorServiceAdapter.forSingleThreadExecutor(
- JM_MAIN_THREAD_EXECUTOR_RESOURCE.getExecutor());
-
+ // Use a direct future executor so the asynchronous deployment
callbacks complete inline
+ // before the manual state transitions below, avoiding the FLINK-38536
race.
final SchedulerBase scheduler =
createScheduler(
JobGraphTestUtils.streamingJobGraph(vertex1, vertex2),
- mainThreadExecutor,
- EXECUTOR_RESOURCE.getExecutor());
+
ComponentMainThreadExecutorServiceAdapter.forMainThread(),
+ new DirectScheduledExecutorService());
+ 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);
-
- runInMainThread(
- mainThreadExecutor, () ->
ExecutionGraphTestUtils.switchAllVerticesToRunning(eg));
+ ExecutionGraphTestUtils.switchAllVerticesToRunning(eg);
// move all vertices to finished state
- runInMainThread(mainThreadExecutor, () ->
ExecutionGraphTestUtils.finishAllVertices(eg));
+ 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(supplyInMainThread(mainThreadExecutor,
eg::getRegisteredExecutions)).isEmpty();
+ assertThat(eg.getRegisteredExecutions()).isEmpty();
}
@Test
@@ -108,54 +82,30 @@ class FinalizeOnMasterTest {
vertex.setInvokableClass(NoOpInvokable.class);
vertex.setParallelism(1);
- final ComponentMainThreadExecutor mainThreadExecutor =
-
ComponentMainThreadExecutorServiceAdapter.forSingleThreadExecutor(
- JM_MAIN_THREAD_EXECUTOR_RESOURCE.getExecutor());
-
+ // Use a direct future executor so the asynchronous deployment
callbacks complete inline
+ // before the manual state transitions below, avoiding the FLINK-38536
race.
final SchedulerBase scheduler =
createScheduler(
JobGraphTestUtils.streamingJobGraph(vertex),
- mainThreadExecutor,
- EXECUTOR_RESOURCE.getExecutor());
+
ComponentMainThreadExecutorServiceAdapter.forMainThread(),
+ new DirectScheduledExecutorService());
+ scheduler.startScheduling();
final ExecutionGraph eg = scheduler.getExecutionGraph();
- runInMainThread(mainThreadExecutor, scheduler::startScheduling);
- assertThat(supplyInMainThread(mainThreadExecutor, eg::getState))
- .isEqualTo(JobStatus.RUNNING);
+ assertThat(eg.getState()).isEqualTo(JobStatus.RUNNING);
- runInMainThread(
- mainThreadExecutor, () ->
ExecutionGraphTestUtils.switchAllVerticesToRunning(eg));
+ ExecutionGraphTestUtils.switchAllVerticesToRunning(eg);
// fail the execution
- runInMainThread(
- mainThreadExecutor,
- () -> {
- final Execution exec =
- eg.getJobVertex(vertex.getID())
- .getTaskVertices()[0]
- .getCurrentExecutionAttempt();
- exec.fail(new Exception("test"));
- });
+ 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(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();
+ assertThat(eg.getRegisteredExecutions()).isEmpty();
}
}
