You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ju...@apache.org on 2024/12/25 02:00:53 UTC
(flink) branch master updated: [FLINK-36921][runtime] Fix unstable AdaptiveExecutionPlanSchedulingContextTest
This is an automated email from the ASF dual-hosted git repository.
junrui 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 1e2787a1eb2 [FLINK-36921][runtime] Fix unstable AdaptiveExecutionPlanSchedulingContextTest
1e2787a1eb2 is described below
commit 1e2787a1eb29ccc8ba2a30d454658c091f7213e0
Author: JunRuiLee <jr...@gmail.com>
AuthorDate: Wed Dec 25 00:13:53 2024 +0800
[FLINK-36921][runtime] Fix unstable AdaptiveExecutionPlanSchedulingContextTest
---
.../AdaptiveExecutionPlanSchedulingContextTest.java | 16 +++++++---------
1 file changed, 7 insertions(+), 9 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveExecutionPlanSchedulingContextTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveExecutionPlanSchedulingContextTest.java
index 14f340cb3b2..500d65b67ff 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveExecutionPlanSchedulingContextTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptivebatch/AdaptiveExecutionPlanSchedulingContextTest.java
@@ -32,7 +32,6 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.RegisterExtension;
import java.util.Collections;
-import java.util.Iterator;
import java.util.concurrent.ScheduledExecutorService;
import static org.apache.flink.runtime.scheduler.SchedulerBase.getDefaultMaxParallelism;
@@ -142,15 +141,14 @@ class AdaptiveExecutionPlanSchedulingContextTest {
env.fromSequence(0L, 1L).disableChaining().print();
StreamGraph streamGraph = env.getStreamGraph();
- Iterator<StreamNode> iterator = streamGraph.getStreamNodes().iterator();
+ for (StreamNode streamNode : streamGraph.getStreamNodes()) {
+ if (streamNode.getOperatorName().contains("Sink")) {
+ streamNode.setParallelism(sinkParallelism);
- iterator.next();
- StreamNode sink = iterator.next();
-
- sink.setParallelism(sinkParallelism);
-
- if (sinkMaxParallelism > 0) {
- sink.setMaxParallelism(sinkMaxParallelism);
+ if (sinkMaxParallelism > 0) {
+ streamNode.setMaxParallelism(sinkMaxParallelism);
+ }
+ }
}
return new DefaultAdaptiveExecutionHandler(