(flink) branch master updated: [FLINK-36921][runtime] Fix unstable AdaptiveExecutionPlanSchedulingContextTest

[junrui]

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 <jrlee....@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(