You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ma...@apache.org on 2026/07/09 12:11:55 UTC
(flink) branch release-2.2 updated (1c7d62e404d -> 82c1e4eb6d1)
This is an automated email from the ASF dual-hosted git repository.
MartijnVisser pushed a change to branch release-2.2
in repository https://gitbox.apache.org/repos/asf/flink.git
from 1c7d62e404d [FLINK-40082][table] Watermark generated method `extractTimestamp` might invoke code throwing exceptions
new 60e6aed8452 [FLINK-39387][Test] Fix flaky scheduler benchmark tests caused by thread assertion failure in async TDD creation
new 6e69f0684f0 [FLINK-39914][runtime] Fix flaky TaskDeploymentDescriptorFactoryTest#testHybridVertexFinish caused by async TDD creation (#28398)
new 17d3f2717fa [FLINK-39922][runtime/test] Fix the flaky test case AbstractAsyncRunnableStreamOperatorTest#testCheckpointDrain (#28425)
new 09377371e95 [FLINK-39921][runtime/tests] Fix the flaky test case ExecutionVertexCancelTest.testSendCancelAndReceiveFail (#28449)
new 82c1e4eb6d1 [FLINK-39929][runtime/tests] Fix the flaky test case ExecutionTimeBasedSlowTaskDetectorTest due to unexepected ComponentMainThreadExecutor setting. (#28434)
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../AbstractAsyncRunnableStreamOperatorTest.java | 3 ++
...MainThreadCheckComponentMainThreadExecutor.java | 40 ++++++++--------
.../TaskDeploymentDescriptorFactoryTest.java | 20 ++++++--
.../executiongraph/ExecutionVertexCancelTest.java | 9 ++--
.../coordination/EventReceivingTasks.java | 53 +---------------------
.../benchmark/SchedulerBenchmarkUtils.java | 4 +-
.../ExecutionTimeBasedSlowTaskDetectorTest.java | 14 ++++--
7 files changed, 60 insertions(+), 83 deletions(-)
copy flink-core/src/main/java/org/apache/flink/util/concurrent/ScheduledExecutorServiceAdapter.java => flink-runtime/src/test/java/org/apache/flink/runtime/concurrent/NoMainThreadCheckComponentMainThreadExecutor.java (56%)
(flink) 02/05: [FLINK-39914][runtime] Fix flaky TaskDeploymentDescriptorFactoryTest#testHybridVertexFinish caused by async TDD creation (#28398)
Posted by ma...@apache.org.
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
commit 6e69f0684f08bf75415769e427f921e9c46cc8be
Author: Chan hae OH <45...@users.noreply.github.com>
AuthorDate: Sun Jun 14 04:10:14 2026 +0900
[FLINK-39914][runtime] Fix flaky TaskDeploymentDescriptorFactoryTest#testHybridVertexFinish caused by async TDD creation (#28398)
(cherry picked from commit 6d47db1d54147eb0da70d5fd39697910ae3aea9f)
---
.../TaskDeploymentDescriptorFactoryTest.java | 20 +++++++++++++++-----
1 file changed, 15 insertions(+), 5 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptorFactoryTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptorFactoryTest.java
index 6960ec3ff67..e34760696c3 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptorFactoryTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/deployment/TaskDeploymentDescriptorFactoryTest.java
@@ -35,6 +35,7 @@ import org.apache.flink.runtime.executiongraph.ExecutionVertex;
import org.apache.flink.runtime.executiongraph.IntermediateResult;
import org.apache.flink.runtime.executiongraph.MarkPartitionFinishedStrategy;
import org.apache.flink.runtime.executiongraph.TestingDefaultExecutionGraphBuilder;
+import org.apache.flink.runtime.executiongraph.utils.ExecutionUtils;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
import org.apache.flink.runtime.jobgraph.IntermediateResultPartitionID;
@@ -49,6 +50,7 @@ import org.apache.flink.runtime.scheduler.SchedulerBase;
import org.apache.flink.runtime.shuffle.ShuffleDescriptor;
import org.apache.flink.testutils.TestingUtils;
import org.apache.flink.testutils.executor.TestExecutorExtension;
+import org.apache.flink.util.Preconditions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
@@ -58,6 +60,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
+import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledExecutorService;
import static org.apache.flink.configuration.JobManagerOptions.HybridPartitionDataConsumeConstraint.ONLY_FINISHED_PRODUCERS;
@@ -221,18 +224,25 @@ class TaskDeploymentDescriptorFactoryTest {
ResultPartitionType.HYBRID_FULL);
JobGraph jobGraph = JobGraphTestUtils.batchJobGraph(producer, consumer);
+
SchedulerBase scheduler =
new DefaultSchedulerBuilder(
jobGraph,
- ComponentMainThreadExecutorServiceAdapter.forMainThread(),
+ ComponentMainThreadExecutorServiceAdapter.forSingleThreadExecutor(
+ EXECUTOR_RESOURCE.getExecutor()),
EXECUTOR_RESOURCE.getExecutor())
.setHybridPartitionDataConsumeConstraint(ONLY_FINISHED_PRODUCERS)
.buildAdaptiveBatchJobScheduler();
- scheduler.startScheduling();
+ CompletableFuture.runAsync(scheduler::startScheduling, EXECUTOR_RESOURCE.getExecutor())
+ .join();
+ // Barrier: ensure the async vertex-initialization action queued inside startScheduling
+ // has completed before we access getTaskVertices().
+ CompletableFuture.runAsync(() -> {}, EXECUTOR_RESOURCE.getExecutor()).join();
ExecutionGraph executionGraph = scheduler.getExecutionGraph();
- return Tuple2.of(
- executionGraph.getJobVertex(producer.getID()),
- executionGraph.getJobVertex(consumer.getID()));
+ ExecutionJobVertex producerVertex =
+ Preconditions.checkNotNull(executionGraph.getJobVertex(producer.getID()));
+ ExecutionUtils.waitForTaskDeploymentDescriptorsCreation(producerVertex.getTaskVertices());
+ return Tuple2.of(producerVertex, executionGraph.getJobVertex(consumer.getID()));
}
// ============== Utils ==============
(flink) 04/05: [FLINK-39921][runtime/tests] Fix the flaky test case ExecutionVertexCancelTest.testSendCancelAndReceiveFail (#28449)
Posted by ma...@apache.org.
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
commit 09377371e9539d5c50ead5b6454cec80483b0dfa
Author: Chan hae OH <45...@users.noreply.github.com>
AuthorDate: Sat Jun 20 12:21:45 2026 +0900
[FLINK-39921][runtime/tests] Fix the flaky test case ExecutionVertexCancelTest.testSendCancelAndReceiveFail (#28449)
(cherry picked from commit a8d6a287ae280fc357d85d7f2efe08c561c8e02e)
---
.../flink/runtime/executiongraph/ExecutionVertexCancelTest.java | 9 ++++++---
1 file changed, 6 insertions(+), 3 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexCancelTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexCancelTest.java
index 35c65ec1763..91f0bdb7e73 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexCancelTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexCancelTest.java
@@ -19,8 +19,9 @@
package org.apache.flink.runtime.executiongraph;
import org.apache.flink.api.common.JobStatus;
-import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutorServiceAdapter;
+import org.apache.flink.runtime.concurrent.NoMainThreadCheckComponentMainThreadExecutor;
import org.apache.flink.runtime.execution.ExecutionState;
+import org.apache.flink.runtime.executiongraph.utils.ExecutionUtils;
import org.apache.flink.runtime.executiongraph.utils.SimpleAckingTaskManagerGateway;
import org.apache.flink.runtime.jobgraph.JobGraphTestUtils;
import org.apache.flink.runtime.jobmaster.LogicalSlot;
@@ -247,12 +248,14 @@ class ExecutionVertexCancelTest {
final SchedulerBase scheduler =
SchedulerTestingUtils.createScheduler(
JobGraphTestUtils.streamingJobGraph(createNoOpVertex(10)),
- ComponentMainThreadExecutorServiceAdapter.forMainThread(),
+ new NoMainThreadCheckComponentMainThreadExecutor(),
EXECUTOR_RESOURCE.getExecutor());
final ExecutionGraph graph = scheduler.getExecutionGraph();
scheduler.startScheduling();
-
+ for (ExecutionVertex ev : graph.getAllExecutionVertices()) {
+ ExecutionUtils.waitForTaskDeploymentDescriptorsCreation(ev);
+ }
ExecutionGraphTestUtils.switchAllVerticesToRunning(graph);
assertThat(graph.getState()).isEqualTo(JobStatus.RUNNING);
(flink) 01/05: [FLINK-39387][Test] Fix flaky scheduler benchmark tests caused by thread assertion failure in async TDD creation
Posted by ma...@apache.org.
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
commit 60e6aed84528f186bdb7977c9e484385f40d0f26
Author: Liu Jiangang <li...@gmail.com>
AuthorDate: Fri Apr 10 10:25:25 2026 +0800
[FLINK-39387][Test] Fix flaky scheduler benchmark tests caused by thread assertion failure in async TDD creation
(cherry picked from commit 106acbced18b0da5fa17181ef6a72738e5ab211c)
---
...MainThreadCheckComponentMainThreadExecutor.java | 70 ++++++++++++++++++++++
.../coordination/EventReceivingTasks.java | 53 +---------------
.../benchmark/SchedulerBenchmarkUtils.java | 4 +-
3 files changed, 73 insertions(+), 54 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/concurrent/NoMainThreadCheckComponentMainThreadExecutor.java b/flink-runtime/src/test/java/org/apache/flink/runtime/concurrent/NoMainThreadCheckComponentMainThreadExecutor.java
new file mode 100644
index 00000000000..894f6365206
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/concurrent/NoMainThreadCheckComponentMainThreadExecutor.java
@@ -0,0 +1,70 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.concurrent;
+
+import org.apache.flink.runtime.testutils.DirectScheduledExecutorService;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * A synchronous {@link ComponentMainThreadExecutor} that executes tasks directly on the calling
+ * thread without performing strict thread identity checks.
+ *
+ * <p>Unlike {@link ComponentMainThreadExecutorServiceAdapter#forMainThread()}, this executor does
+ * not assert that the current thread is the main thread, avoiding flaky test failures when {@link
+ * java.util.concurrent.CompletableFuture} callbacks are dispatched from background threads.
+ */
+public class NoMainThreadCheckComponentMainThreadExecutor implements ComponentMainThreadExecutor {
+
+ private final DirectScheduledExecutorService executor = new DirectScheduledExecutorService();
+
+ @Override
+ public void assertRunningInMainThread() {
+ // No-op: Skip thread assertion to avoid flaky test failures
+ }
+
+ @Override
+ public void execute(Runnable command) {
+ executor.execute(command);
+ }
+
+ @Override
+ public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
+ return executor.schedule(command, delay, unit);
+ }
+
+ @Override
+ public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
+ return executor.schedule(callable, delay, unit);
+ }
+
+ @Override
+ public ScheduledFuture<?> scheduleAtFixedRate(
+ Runnable command, long initialDelay, long period, TimeUnit unit) {
+ return executor.scheduleAtFixedRate(command, initialDelay, period, unit);
+ }
+
+ @Override
+ public ScheduledFuture<?> scheduleWithFixedDelay(
+ Runnable command, long initialDelay, long delay, TimeUnit unit) {
+ return executor.scheduleWithFixedDelay(command, initialDelay, delay, unit);
+ }
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/EventReceivingTasks.java b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/EventReceivingTasks.java
index aa27bb2f907..2ceddc7176b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/EventReceivingTasks.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/operators/coordination/EventReceivingTasks.java
@@ -18,18 +18,13 @@
package org.apache.flink.runtime.operators.coordination;
-import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
+import org.apache.flink.runtime.concurrent.NoMainThreadCheckComponentMainThreadExecutor;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.runtime.operators.coordination.util.IncompleteFuturesTracker;
-import org.apache.flink.runtime.testutils.DirectScheduledExecutorService;
import org.apache.flink.util.SerializedValue;
import org.apache.flink.util.concurrent.FutureUtils;
-import org.apache.flink.util.concurrent.ScheduledExecutor;
-import org.apache.flink.util.concurrent.ScheduledExecutorServiceAdapter;
-
-import javax.annotation.Nonnull;
import java.io.IOException;
import java.util.ArrayList;
@@ -39,8 +34,6 @@ import java.util.List;
import java.util.Objects;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ScheduledFuture;
-import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import static org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils.createExecutionAttemptId;
@@ -242,48 +235,4 @@ public class EventReceivingTasks implements SubtaskAccess.SubtaskAccessFactory {
return taskFailoverReasons;
}
}
-
- /**
- * An implementation of {@link ComponentMainThreadExecutor} that executes Runnables with a
- * wrapped {@link ScheduledExecutor} and disables {@link #assertRunningInMainThread()} checks.
- */
- private static class NoMainThreadCheckComponentMainThreadExecutor
- implements ComponentMainThreadExecutor {
- private final ScheduledExecutor scheduledExecutor;
-
- private NoMainThreadCheckComponentMainThreadExecutor() {
- this.scheduledExecutor =
- new ScheduledExecutorServiceAdapter(new DirectScheduledExecutorService());
- }
-
- @Override
- public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
- return scheduledExecutor.schedule(command, delay, unit);
- }
-
- @Override
- public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
- return scheduledExecutor.schedule(callable, delay, unit);
- }
-
- @Override
- public ScheduledFuture<?> scheduleAtFixedRate(
- Runnable command, long initialDelay, long period, TimeUnit unit) {
- return scheduledExecutor.scheduleAtFixedRate(command, initialDelay, period, unit);
- }
-
- @Override
- public ScheduledFuture<?> scheduleWithFixedDelay(
- Runnable command, long initialDelay, long delay, TimeUnit unit) {
- return scheduledExecutor.scheduleAtFixedRate(command, initialDelay, delay, unit);
- }
-
- @Override
- public void assertRunningInMainThread() {}
-
- @Override
- public void execute(@Nonnull Runnable command) {
- scheduledExecutor.execute(command);
- }
- }
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/SchedulerBenchmarkUtils.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/SchedulerBenchmarkUtils.java
index baab5227f09..988f90b6d57 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/SchedulerBenchmarkUtils.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/benchmark/SchedulerBenchmarkUtils.java
@@ -21,7 +21,7 @@ package org.apache.flink.runtime.scheduler.benchmark;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.runtime.JobException;
import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
-import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutorServiceAdapter;
+import org.apache.flink.runtime.concurrent.NoMainThreadCheckComponentMainThreadExecutor;
import org.apache.flink.runtime.execution.ExecutionState;
import org.apache.flink.runtime.executiongraph.AccessExecutionJobVertex;
import org.apache.flink.runtime.executiongraph.Execution;
@@ -111,7 +111,7 @@ public class SchedulerBenchmarkUtils {
final JobGraph jobGraph = createJobGraph(jobVertices, jobConfiguration);
final ComponentMainThreadExecutor mainThreadExecutor =
- ComponentMainThreadExecutorServiceAdapter.forMainThread();
+ new NoMainThreadCheckComponentMainThreadExecutor();
DefaultSchedulerBuilder schedulerBuilder =
new DefaultSchedulerBuilder(jobGraph, mainThreadExecutor, scheduledExecutorService)
(flink) 05/05: [FLINK-39929][runtime/tests] Fix the flaky test case ExecutionTimeBasedSlowTaskDetectorTest due to unexepected ComponentMainThreadExecutor setting. (#28434)
Posted by ma...@apache.org.
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
commit 82c1e4eb6d1a2d7dc296cca1ac39b60cd7bae076
Author: Chan hae OH <45...@users.noreply.github.com>
AuthorDate: Wed Jun 24 14:05:38 2026 +0900
[FLINK-39929][runtime/tests] Fix the flaky test case ExecutionTimeBasedSlowTaskDetectorTest due to unexepected ComponentMainThreadExecutor setting. (#28434)
Co-authored-by: Yuepeng Pan <hi...@gmail.com>
(cherry picked from commit 67eba4581147ac527fb7804687ca24745dde08da)
---
.../ExecutionTimeBasedSlowTaskDetectorTest.java | 14 +++++++++++---
1 file changed, 11 insertions(+), 3 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/slowtaskdetector/ExecutionTimeBasedSlowTaskDetectorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/slowtaskdetector/ExecutionTimeBasedSlowTaskDetectorTest.java
index 5ec9baeb3cc..d01a095a66c 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/slowtaskdetector/ExecutionTimeBasedSlowTaskDetectorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/slowtaskdetector/ExecutionTimeBasedSlowTaskDetectorTest.java
@@ -22,10 +22,12 @@ package org.apache.flink.runtime.scheduler.slowtaskdetector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.SlowTaskDetectorOptions;
import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutorServiceAdapter;
+import org.apache.flink.runtime.concurrent.NoMainThreadCheckComponentMainThreadExecutor;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.executiongraph.ExecutionGraph;
import org.apache.flink.runtime.executiongraph.ExecutionGraphTestUtils;
import org.apache.flink.runtime.executiongraph.ExecutionVertex;
+import org.apache.flink.runtime.executiongraph.utils.ExecutionUtils;
import org.apache.flink.runtime.io.network.partition.ResultPartitionType;
import org.apache.flink.runtime.jobgraph.DistributionPattern;
import org.apache.flink.runtime.jobgraph.JobGraph;
@@ -91,7 +93,7 @@ class ExecutionTimeBasedSlowTaskDetectorTest {
final ExecutionGraph executionGraph =
SchedulerTestingUtils.createScheduler(
jobGraph,
- ComponentMainThreadExecutorServiceAdapter.forMainThread(),
+ new NoMainThreadCheckComponentMainThreadExecutor(),
EXECUTOR_RESOURCE.getExecutor())
.getExecutionGraph();
@@ -433,12 +435,15 @@ class ExecutionTimeBasedSlowTaskDetectorTest {
final SchedulerBase scheduler =
SchedulerTestingUtils.createScheduler(
jobGraph,
- ComponentMainThreadExecutorServiceAdapter.forMainThread(),
+ new NoMainThreadCheckComponentMainThreadExecutor(),
EXECUTOR_RESOURCE.getExecutor());
final ExecutionGraph executionGraph = scheduler.getExecutionGraph();
scheduler.startScheduling();
+ for (ExecutionVertex ev : executionGraph.getAllExecutionVertices()) {
+ ExecutionUtils.waitForTaskDeploymentDescriptorsCreation(ev);
+ }
ExecutionGraphTestUtils.switchAllVerticesToRunning(executionGraph);
return executionGraph;
@@ -450,13 +455,16 @@ class ExecutionTimeBasedSlowTaskDetectorTest {
final SchedulerBase scheduler =
new DefaultSchedulerBuilder(
jobGraph,
- ComponentMainThreadExecutorServiceAdapter.forMainThread(),
+ new NoMainThreadCheckComponentMainThreadExecutor(),
EXECUTOR_RESOURCE.getExecutor())
.buildAdaptiveBatchJobScheduler();
final ExecutionGraph executionGraph = scheduler.getExecutionGraph();
scheduler.startScheduling();
+ for (ExecutionVertex ev : executionGraph.getAllExecutionVertices()) {
+ ExecutionUtils.waitForTaskDeploymentDescriptorsCreation(ev);
+ }
ExecutionGraphTestUtils.switchAllVerticesToRunning(executionGraph);
return executionGraph;
(flink) 03/05: [FLINK-39922][runtime/test] Fix the flaky test case AbstractAsyncRunnableStreamOperatorTest#testCheckpointDrain (#28425)
Posted by ma...@apache.org.
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
commit 17d3f2717fad8146d62e474adfbfaf811d1b72de
Author: Yuepeng Pan <pa...@apache.org>
AuthorDate: Sun Jun 14 10:16:59 2026 +0800
[FLINK-39922][runtime/test] Fix the flaky test case AbstractAsyncRunnableStreamOperatorTest#testCheckpointDrain (#28425)
(cherry picked from commit 6d106cda908ab1e126619e4898371b0bb9e1c393)
---
.../operators/AbstractAsyncRunnableStreamOperatorTest.java | 3 +++
1 file changed, 3 insertions(+)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncRunnableStreamOperatorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncRunnableStreamOperatorTest.java
index dedf01de245..b073f6cbed1 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncRunnableStreamOperatorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/operators/AbstractAsyncRunnableStreamOperatorTest.java
@@ -241,17 +241,20 @@ public class AbstractAsyncRunnableStreamOperatorTest {
(SimpleAsyncExecutionController)
((AbstractAsyncRunnableStreamOperator) testHarness.getOperator())
.getAsyncExecutionController();
+ CompletableFuture<Void> unblockAsyncRequest = new CompletableFuture<>();
((AbstractAsyncRunnableStreamOperator<String>) testHarness.getOperator())
.setAsyncKeyedContextElement(
new StreamRecord<>(Tuple2.of(5, "5")), new TestKeySelector());
((AbstractAsyncRunnableStreamOperator<String>) testHarness.getOperator())
.asyncProcess(
() -> {
+ unblockAsyncRequest.get(10, TimeUnit.SECONDS);
return null;
});
((AbstractAsyncRunnableStreamOperator<String>) testHarness.getOperator())
.postProcessElement();
assertThat(asyncExecutionController.getInFlightRecordNum()).isEqualTo(1);
+ unblockAsyncRequest.complete(null);
testHarness.drainAsyncRequests();
assertThat(asyncExecutionController.getInFlightRecordNum()).isEqualTo(0);
}