(flink) branch master updated: [FLINK-35095][test] Fix unstable tests in `ExecutionEnvironmentImplTest`

[guoweijie]

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

guoweijie 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 fd4b4005c93 [FLINK-35095][test] Fix unstable tests in 
`ExecutionEnvironmentImplTest`
fd4b4005c93 is described below

commit fd4b4005c932ac77c693e6a691878407da8377d7
Author: Weijie Guo <res...@163.com>
AuthorDate: Mon Jul 15 11:57:21 2024 +0800

    [FLINK-35095][test] Fix unstable tests in `ExecutionEnvironmentImplTest`
---
 .../datastream/impl/ExecutionEnvironmentImplTest.java      | 14 ++++++++++----
 1 file changed, 10 insertions(+), 4 deletions(-)

diff --git 
a/flink-datastream/src/test/java/org/apache/flink/datastream/impl/ExecutionEnvironmentImplTest.java
 
b/flink-datastream/src/test/java/org/apache/flink/datastream/impl/ExecutionEnvironmentImplTest.java
index 137429ebcf8..4ab33e12df5 100644
--- 
a/flink-datastream/src/test/java/org/apache/flink/datastream/impl/ExecutionEnvironmentImplTest.java
+++ 
b/flink-datastream/src/test/java/org/apache/flink/datastream/impl/ExecutionEnvironmentImplTest.java
@@ -93,8 +93,11 @@ class ExecutionEnvironmentImplTest {
         Collection<StreamNode> nodes = streamGraph.getStreamNodes();
         assertThat(nodes).hasSize(2);
         Collection<Integer> sourceIDs = streamGraph.getSourceIDs();
-        StreamNode sourceNode = nodes.iterator().next();
-        assertThat(sourceIDs).containsExactly(sourceNode.getId());
+        for (StreamNode node : nodes) {
+            if (node.getOperatorName().contains("source")) {
+                assertThat(sourceIDs).containsExactly(node.getId());
+            }
+        }
     }
 
     @Test
@@ -118,7 +121,10 @@ class ExecutionEnvironmentImplTest {
         StreamGraph streamGraph = StreamTestUtils.getStreamGraph(env);
         Collection<StreamNode> nodes = streamGraph.getStreamNodes();
         Collection<Integer> sourceIDs = streamGraph.getSourceIDs();
-        StreamNode sourceNode = nodes.iterator().next();
-        assertThat(sourceIDs).containsExactly(sourceNode.getId());
+        for (StreamNode node : nodes) {
+            if (node.getOperatorName().contains("source")) {
+                assertThat(sourceIDs).containsExactly(node.getId());
+            }
+        }
     }
 }