diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationGroup.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationGroup.java index 2b606d6a22859e..e42be651656916 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationGroup.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationGroup.java @@ -52,7 +52,7 @@ public void addVertex(JobVertex vertex) { } @Override - public List getVertices() { + public List getVertexIDs() { return vertices.stream().map(JobVertex::getID).collect(ImmutableList.toImmutableList()); } diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationGroupDesc.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationGroupDesc.java index 6507182dd64131..e0fa9e2ae03e6f 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationGroupDesc.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationGroupDesc.java @@ -46,7 +46,7 @@ public interface CoLocationGroupDesc { * * @return The group's members represented by their {@link JobVertexID}s. */ - List getVertices(); + List getVertexIDs(); /** * Returns the {@link CoLocationConstraintDesc} for a specific {@code subTaskIndex}. diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/LocalInputPreferredSlotSharingStrategy.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/LocalInputPreferredSlotSharingStrategy.java index 5c0ef271272d89..172025703f4dd8 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/LocalInputPreferredSlotSharingStrategy.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/LocalInputPreferredSlotSharingStrategy.java @@ -117,7 +117,7 @@ private ExecutionSlotSharingGroupBuilder( this.coLocationGroupMap = new HashMap<>(); for (CoLocationGroupDesc coLocationGroup : coLocationGroups) { - for (JobVertexID jobVertexId : coLocationGroup.getVertices()) { + for (JobVertexID jobVertexId : coLocationGroup.getVertexIDs()) { coLocationGroupMap.put(jobVertexId, coLocationGroup); } } diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/JobGraphTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/JobGraphTest.java index 763daceb02276b..7c569ed6201565 100644 --- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/JobGraphTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobgraph/JobGraphTest.java @@ -444,6 +444,6 @@ public void testGetCoLocationGroupDescriptors() { final CoLocationGroupDesc onlyCoLocationGroupDesc = jobGraph.getCoLocationGroupDescriptors().iterator().next(); assertThat( - onlyCoLocationGroupDesc.getVertices(), containsInAnyOrder(v1.getID(), v2.getID())); + onlyCoLocationGroupDesc.getVertexIDs(), containsInAnyOrder(v1.getID(), v2.getID())); } } diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java index 2660011118ed4a..c5f58289db9d70 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java @@ -819,10 +819,10 @@ public void testIteration() { if (jobVertex.getName().startsWith(StreamGraph.ITERATION_SOURCE_NAME_PREFIX)) { iterationSourceCoLocationGroup = jobVertex.getCoLocationGroup(); assertTrue( - iterationSourceCoLocationGroup.getVertices().contains(jobVertex.getID())); + iterationSourceCoLocationGroup.getVertexIDs().contains(jobVertex.getID())); } else if (jobVertex.getName().startsWith(StreamGraph.ITERATION_SINK_NAME_PREFIX)) { iterationSinkCoLocationGroup = jobVertex.getCoLocationGroup(); - assertTrue(iterationSinkCoLocationGroup.getVertices().contains(jobVertex.getID())); + assertTrue(iterationSinkCoLocationGroup.getVertexIDs().contains(jobVertex.getID())); } else { assertNull(jobVertex.getCoLocationGroup()); }