(flink) branch release-2.2 updated: [FLINK-38908][flink-runtime][test] Fix flaky ExecutionPartitionLifecycleTest

[martijnvisser]

This is an automated email from the ASF dual-hosted git repository.

MartijnVisser pushed a commit to branch release-2.2
in repository https://gitbox.apache.org/repos/asf/flink.git


The following commit(s) were added to refs/heads/release-2.2 by this push:
     new 685e443d072 [FLINK-38908][flink-runtime][test] Fix flaky 
ExecutionPartitionLifecycleTest
685e443d072 is described below

commit 685e443d0722426a822e11498b7a930736321896
Author: Mate Czagany <4469996+mateczag...@users.noreply.github.com>
AuthorDate: Wed Jan 28 12:50:54 2026 +0100

    [FLINK-38908][flink-runtime][test] Fix flaky ExecutionPartitionLifecycleTest
    
    (cherry picked from commit 76fbab4f20c476a8f3635cc9df2971ed98a249d2)
---
 .../ExecutionPartitionLifecycleTest.java           | 39 ++++++++++++++++------
 1 file changed, 28 insertions(+), 11 deletions(-)

diff --git 
a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionPartitionLifecycleTest.java
 
b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionPartitionLifecycleTest.java
index c20087155ec..1b4d8db7815 100644
--- 
a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionPartitionLifecycleTest.java
+++ 
b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionPartitionLifecycleTest.java
@@ -23,6 +23,7 @@ import org.apache.flink.api.java.tuple.Tuple2;
 import org.apache.flink.runtime.clusterframework.types.ResourceID;
 import 
org.apache.flink.runtime.concurrent.ComponentMainThreadExecutorServiceAdapter;
 import org.apache.flink.runtime.deployment.ResultPartitionDeploymentDescriptor;
+import org.apache.flink.runtime.executiongraph.utils.ExecutionUtils;
 import 
org.apache.flink.runtime.executiongraph.utils.SimpleAckingTaskManagerGateway;
 import org.apache.flink.runtime.io.network.partition.JobMasterPartitionTracker;
 import 
org.apache.flink.runtime.io.network.partition.NoOpJobMasterPartitionTracker;
@@ -74,7 +75,11 @@ class ExecutionPartitionLifecycleTest {
 
     @RegisterExtension
     static final TestExecutorExtension<ScheduledExecutorService> 
EXECUTOR_RESOURCE =
-            TestingUtils.defaultExecutorExtension();
+            TestingUtils.jmAsyncThreadExecutorExtension();
+
+    @RegisterExtension
+    static final TestExecutorExtension<ScheduledExecutorService> 
JM_MAIN_THREAD_EXECUTOR_RESOURCE =
+            TestingUtils.jmMainThreadExecutorExtension();
 
     private Execution execution;
     private ResultPartitionDeploymentDescriptor descriptor;
@@ -118,11 +123,14 @@ class ExecutionPartitionLifecycleTest {
                 taskManagerGateway,
                 testingShuffleMaster);
 
-        stateTransition1.accept(execution);
-        assertThat(releasePartitionsCallFuture).isNotDone();
+        runInMainThread(
+                () -> {
+                    stateTransition1.accept(execution);
+                    assertThat(releasePartitionsCallFuture).isNotDone();
 
-        stateTransition2.accept(execution);
-        assertThat(releasePartitionsCallFuture).isDone();
+                    stateTransition2.accept(execution);
+                    assertThat(releasePartitionsCallFuture).isDone();
+                });
 
         final Tuple2<JobID, Collection<ResultPartitionID>> 
releasePartitionsCall =
                 releasePartitionsCallFuture.get();
@@ -221,7 +229,10 @@ class ExecutionPartitionLifecycleTest {
         assertThat(startTrackingCall.f0).isEqualTo(taskExecutorResourceId);
         assertThat(startTrackingCall.f1).isEqualTo(descriptor);
 
-        stateTransition.accept(execution);
+        runInMainThread(
+                () -> {
+                    stateTransition.accept(execution);
+                });
 
         switch (partitionReleaseResult) {
             case NONE:
@@ -279,7 +290,8 @@ class ExecutionPartitionLifecycleTest {
         final SchedulerBase scheduler =
                 new DefaultSchedulerBuilder(
                                 jobGraph,
-                                
ComponentMainThreadExecutorServiceAdapter.forMainThread(),
+                                
ComponentMainThreadExecutorServiceAdapter.forSingleThreadExecutor(
+                                        
JM_MAIN_THREAD_EXECUTOR_RESOURCE.getExecutor()),
                                 EXECUTOR_RESOURCE.getExecutor())
                         .setExecutionSlotAllocatorFactory(
                                 
SchedulerTestingUtils.newSlotSharingExecutionSlotAllocatorFactory(
@@ -289,15 +301,16 @@ class ExecutionPartitionLifecycleTest {
                         .build();
 
         final ExecutionGraph executionGraph = scheduler.getExecutionGraph();
-
         final ExecutionJobVertex executionJobVertex =
                 executionGraph.getJobVertex(producerVertex.getID());
         final ExecutionVertex executionVertex = 
executionJobVertex.getTaskVertices()[0];
         execution = executionVertex.getCurrentExecutionAttempt();
 
-        scheduler.startScheduling();
-        execution.switchToInitializing();
-        execution.switchToRunning();
+        runInMainThread(scheduler::startScheduling);
+        
ExecutionUtils.waitForTaskDeploymentDescriptorsCreation(executionVertex);
+
+        assertThat(execution.switchToInitializing()).isTrue();
+        assertThat(execution.switchToRunning()).isTrue();
 
         final IntermediateResultPartitionID 
expectedIntermediateResultPartitionId =
                 
executionJobVertex.getProducedDataSets()[0].getPartitions()[0].getPartitionId();
@@ -319,6 +332,10 @@ class ExecutionPartitionLifecycleTest {
         return jobVertex;
     }
 
+    private static void runInMainThread(final Runnable runnable) {
+        CompletableFuture.runAsync(runnable, 
JM_MAIN_THREAD_EXECUTOR_RESOURCE.getExecutor()).join();
+    }
+
     private static class TestingShuffleMaster implements 
ShuffleMaster<ShuffleDescriptor> {
 
         final Queue<ShuffleDescriptor> externallyReleasedPartitions = new 
ArrayBlockingQueue<>(4);