[FLINK-20850][runtime] Apply renamings

Minor renamings are applied that were brought up during the review. Additionally, missing JavaDoc was added.
apache · Jan 12, 2021 · 51e4355 · 51e4355
1 parent 7dea123
commit 51e4355
Show file tree

Hide file tree

Showing 8 changed files with 18 additions and 12 deletions.
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/JobGraph.java
@@ -331,7 +331,12 @@ public Set<SlotSharingGroup> getSlotSharingGroups() {
         return Collections.unmodifiableSet(slotSharingGroups);
     }
 
-    public Set<CoLocationGroup> getCoLocationGroup() {
+    /**
+     * Returns all {@link CoLocationGroup} instances associated with this {@code JobGraph}.
+     *
+     * @return The associated {@code CoLocationGroup} instances.
+     */
+    public Set<CoLocationGroup> getCoLocationGroups() {
         final Set<CoLocationGroup> coLocationGroups =
                 IterableUtils.toStream(getVertices())
                         .map(JobVertex::getCoLocationGroup)

diff --git a/...-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationGroup.java b/...-runtime/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationGroup.java
@@ -46,7 +46,7 @@ public interface CoLocationGroup {
      *
      * @return The group's members represented by their {@link JobVertexID}s.
      */
-    List<JobVertexID> getVertexIDs();
+    List<JobVertexID> getVertexIds();
 
     /**
      * Returns the {@link CoLocationConstraint} for a specific {@code subTaskIndex}.

diff --git a/...time/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationGroupImpl.java b/...time/src/main/java/org/apache/flink/runtime/jobmanager/scheduler/CoLocationGroupImpl.java
@@ -52,7 +52,7 @@ public void addVertex(JobVertex vertex) {
     }
 
     @Override
-    public List<JobVertexID> getVertexIDs() {
+    public List<JobVertexID> getVertexIds() {
         return vertices.stream().map(JobVertex::getID).collect(ImmutableList.toImmutableList());
     }
 

diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultScheduler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultScheduler.java
@@ -600,7 +600,7 @@ public Set<SlotSharingGroup> getLogicalSlotSharingGroups() {
 
         @Override
         public Set<CoLocationGroup> getCoLocationGroups() {
-            return getJobGraph().getCoLocationGroup();
+            return getJobGraph().getCoLocationGroups();
         }
 
         @Override

diff --git a/.../main/java/org/apache/flink/runtime/scheduler/LocalInputPreferredSlotSharingStrategy.java b/.../main/java/org/apache/flink/runtime/scheduler/LocalInputPreferredSlotSharingStrategy.java
@@ -117,7 +117,7 @@ private ExecutionSlotSharingGroupBuilder(
 
             this.coLocationGroupMap = new HashMap<>();
             for (CoLocationGroup coLocationGroup : coLocationGroups) {
-                for (JobVertexID jobVertexId : coLocationGroup.getVertexIDs()) {
+                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
@@ -426,7 +426,7 @@ public void testGetSlotSharingGroups() {
     }
 
     @Test
-    public void testGetCoLocationGroup() {
+    public void testGetCoLocationGroups() {
         final JobVertex v1 = new JobVertex("1");
         final JobVertex v2 = new JobVertex("2");
         final JobVertex v3 = new JobVertex("3");
@@ -439,9 +439,10 @@ public void testGetCoLocationGroup() {
 
         final JobGraph jobGraph = new JobGraph(v1, v2, v3, v4);
 
-        assertThat(jobGraph.getCoLocationGroup(), hasSize(1));
+        assertThat(jobGraph.getCoLocationGroups(), hasSize(1));
 
-        final CoLocationGroup onlyCoLocationGroup = jobGraph.getCoLocationGroup().iterator().next();
-        assertThat(onlyCoLocationGroup.getVertexIDs(), containsInAnyOrder(v1.getID(), v2.getID()));
+        final CoLocationGroup onlyCoLocationGroup =
+                jobGraph.getCoLocationGroups().iterator().next();
+        assertThat(onlyCoLocationGroup.getVertexIds(), containsInAnyOrder(v1.getID(), v2.getID()));
     }
 }
diff --git a/...t/java/org/apache/flink/runtime/scheduler/LocalInputPreferredSlotSharingStrategyTest.java b/...t/java/org/apache/flink/runtime/scheduler/LocalInputPreferredSlotSharingStrategyTest.java
@@ -183,7 +183,7 @@ private TestingCoLocationGroup(JobVertexID... vertexIDs) {
         }
 
         @Override
-        public List<JobVertexID> getVertexIDs() {
+        public List<JobVertexID> getVertexIds() {
             return vertexIDs;
         }
     }

diff --git a/...va/src/test/java/org/apache/flink/streaming/api/graph/StreamingJobGraphGeneratorTest.java b/...va/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.getVertexIDs().contains(jobVertex.getID()));
+                        iterationSourceCoLocationGroup.getVertexIds().contains(jobVertex.getID()));
             } else if (jobVertex.getName().startsWith(StreamGraph.ITERATION_SINK_NAME_PREFIX)) {
                 iterationSinkCoLocationGroup = jobVertex.getCoLocationGroup();
-                assertTrue(iterationSinkCoLocationGroup.getVertexIDs().contains(jobVertex.getID()));
+                assertTrue(iterationSinkCoLocationGroup.getVertexIds().contains(jobVertex.getID()));
             } else {
                 assertNull(jobVertex.getCoLocationGroup());
             }