azagrebin commented on a change in pull request #12423:
URL: https://github.com/apache/flink/pull/12423#discussion_r434727115
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/AbstractExecutionSlotAllocator.java
##########
@@ -105,22 +104,11 @@ protected SlotExecutionVertexAssignment
createAndRegisterSlotExecutionVertexAssi
return slotExecutionVertexAssignment;
}
- /**
- * Calculates the preferred locations for an execution.
- * It will first try to use preferred locations based on state,
- * if null, will use the preferred locations based on inputs.
- */
- protected static CompletableFuture<Collection<TaskManagerLocation>>
calculatePreferredLocations(
+ protected CompletableFuture<Collection<TaskManagerLocation>>
calculatePreferredLocations(
Review comment:
do we need this method `calculatePreferredLocations`?
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/PreferredLocationsRetriever.java
##########
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.scheduler;
+
+import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
+import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
+
+import java.util.Collection;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * Component to retrieve the preferred locations of an execution vertex.
+ */
+public interface PreferredLocationsRetriever {
Review comment:
```suggestion
@FunctionalInterface
public interface PreferredLocationsRetriever {
```
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultPreferredLocationsRetriever.java
##########
@@ -0,0 +1,137 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.scheduler;
+
+import org.apache.flink.runtime.concurrent.FutureUtils;
+import org.apache.flink.runtime.executiongraph.ExecutionVertex;
+import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
+import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+import static
org.apache.flink.runtime.executiongraph.ExecutionVertex.MAX_DISTINCT_LOCATIONS_TO_CONSIDER;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Default implementation of {@link PreferredLocationsRetriever}.
+ * Locations based on state will be returned if exist.
+ * Otherwise locations based on inputs will be returned.
+ */
+public class DefaultPreferredLocationsRetriever implements
PreferredLocationsRetriever {
+
+ private final StateLocationRetriever stateLocationRetriever;
+
+ private final InputsLocationsRetriever inputsLocationsRetriever;
+
+ DefaultPreferredLocationsRetriever(
+ final StateLocationRetriever stateLocationRetriever,
+ final InputsLocationsRetriever
inputsLocationsRetriever) {
+
+ this.stateLocationRetriever =
checkNotNull(stateLocationRetriever);
+ this.inputsLocationsRetriever =
checkNotNull(inputsLocationsRetriever);
+ }
+
+ @Override
+ public CompletableFuture<Collection<TaskManagerLocation>>
getPreferredLocations(
+ final ExecutionVertexID executionVertexId,
+ final Set<ExecutionVertexID> producersToIgnore) {
+
+ checkNotNull(executionVertexId);
+ checkNotNull(producersToIgnore);
+
+ final Collection<TaskManagerLocation>
preferredLocationsBasedOnState =
+ getPreferredLocationsBasedOnState(executionVertexId,
stateLocationRetriever);
+ if (!preferredLocationsBasedOnState.isEmpty()) {
+ return
CompletableFuture.completedFuture(preferredLocationsBasedOnState);
+ }
+
+ return getPreferredLocationsBasedOnInputs(executionVertexId,
producersToIgnore, inputsLocationsRetriever);
+ }
+
+ private static Collection<TaskManagerLocation>
getPreferredLocationsBasedOnState(
+ final ExecutionVertexID executionVertexId,
+ final StateLocationRetriever stateLocationRetriever) {
+
+ return
stateLocationRetriever.getStateLocation(executionVertexId)
+ .map(Collections::singleton)
+ .orElse(Collections.emptySet());
+ }
+
+ /**
+ * Gets the location preferences of the execution, as determined by the
locations
+ * of the predecessors from which it receives input data.
+ * If there are more than {@link
ExecutionVertex#MAX_DISTINCT_LOCATIONS_TO_CONSIDER} different locations of
source data,
+ * or neither the sources have not been started nor will be started
with the execution together,
+ * this method returns an empty collection to indicate no location
preference.
+ *
+ * @return The preferred locations based in input streams, or an empty
iterable,
+ * if there is no input-based preference.
+ */
+ static CompletableFuture<Collection<TaskManagerLocation>>
getPreferredLocationsBasedOnInputs(
Review comment:
I do not quite understand why we need this static, not private method
`getPreferredLocationsBasedOnInputs ` at the end.
Could tests not be written against public interface method?
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerBase.java
##########
@@ -229,7 +229,9 @@ public SchedulerBase(
this.executionGraph =
createAndRestoreExecutionGraph(jobManagerJobMetricGroup,
checkNotNull(shuffleMaster), checkNotNull(partitionTracker));
this.schedulingTopology =
executionGraph.getSchedulingTopology();
- this.inputsLocationsRetriever = new
ExecutionGraphToInputsLocationsRetrieverAdapter(executionGraph);
+ final StateLocationRetriever stateLocationRetriever = new
ExecutionVertexStateLocationRetriever(this::getExecutionVertex);
Review comment:
Maybe, just:
```suggestion
final StateLocationRetriever stateLocationRetriever =
executionVertexId ->
getExecutionVertex(executionVertexId).getPreferredLocationBasedOnState();
```
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultPreferredLocationsRetriever.java
##########
@@ -0,0 +1,137 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.scheduler;
+
+import org.apache.flink.runtime.concurrent.FutureUtils;
+import org.apache.flink.runtime.executiongraph.ExecutionVertex;
+import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
+import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+import static
org.apache.flink.runtime.executiongraph.ExecutionVertex.MAX_DISTINCT_LOCATIONS_TO_CONSIDER;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Default implementation of {@link PreferredLocationsRetriever}.
+ * Locations based on state will be returned if exist.
+ * Otherwise locations based on inputs will be returned.
+ */
+public class DefaultPreferredLocationsRetriever implements
PreferredLocationsRetriever {
+
+ private final StateLocationRetriever stateLocationRetriever;
+
+ private final InputsLocationsRetriever inputsLocationsRetriever;
+
+ DefaultPreferredLocationsRetriever(
+ final StateLocationRetriever stateLocationRetriever,
+ final InputsLocationsRetriever
inputsLocationsRetriever) {
+
+ this.stateLocationRetriever =
checkNotNull(stateLocationRetriever);
+ this.inputsLocationsRetriever =
checkNotNull(inputsLocationsRetriever);
+ }
+
+ @Override
+ public CompletableFuture<Collection<TaskManagerLocation>>
getPreferredLocations(
+ final ExecutionVertexID executionVertexId,
+ final Set<ExecutionVertexID> producersToIgnore) {
+
+ checkNotNull(executionVertexId);
+ checkNotNull(producersToIgnore);
+
+ final Collection<TaskManagerLocation>
preferredLocationsBasedOnState =
+ getPreferredLocationsBasedOnState(executionVertexId,
stateLocationRetriever);
+ if (!preferredLocationsBasedOnState.isEmpty()) {
+ return
CompletableFuture.completedFuture(preferredLocationsBasedOnState);
+ }
+
+ return getPreferredLocationsBasedOnInputs(executionVertexId,
producersToIgnore, inputsLocationsRetriever);
+ }
+
+ private static Collection<TaskManagerLocation>
getPreferredLocationsBasedOnState(
+ final ExecutionVertexID executionVertexId,
+ final StateLocationRetriever stateLocationRetriever) {
+
+ return
stateLocationRetriever.getStateLocation(executionVertexId)
+ .map(Collections::singleton)
+ .orElse(Collections.emptySet());
+ }
+
+ /**
+ * Gets the location preferences of the execution, as determined by the
locations
+ * of the predecessors from which it receives input data.
+ * If there are more than {@link
ExecutionVertex#MAX_DISTINCT_LOCATIONS_TO_CONSIDER} different locations of
source data,
+ * or neither the sources have not been started nor will be started
with the execution together,
+ * this method returns an empty collection to indicate no location
preference.
+ *
+ * @return The preferred locations based in input streams, or an empty
iterable,
+ * if there is no input-based preference.
+ */
+ static CompletableFuture<Collection<TaskManagerLocation>>
getPreferredLocationsBasedOnInputs(
+ final ExecutionVertexID executionVertexId,
+ final Set<ExecutionVertexID> producersToIgnore,
+ final InputsLocationsRetriever
inputsLocationsRetriever) {
+
+ CompletableFuture<Collection<TaskManagerLocation>>
preferredLocations =
+
CompletableFuture.completedFuture(Collections.emptyList());
+
+ final Collection<CompletableFuture<TaskManagerLocation>>
locationsFutures = new ArrayList<>();
Review comment:
we clear this collection at the end of each iteration.
maybe it should go inside?
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/ExecutionVertexSchedulingRequirements.java
##########
@@ -169,8 +161,7 @@ public ExecutionVertexSchedulingRequirements build() {
taskResourceProfile,
physicalSlotResourceProfile,
slotSharingGroupId,
- coLocationConstraint,
- preferredLocations);
+ coLocationConstraint);
Review comment:
`withPreferredLocations` and `this.preferredLocations` can be also
removed
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultPreferredLocationsRetriever.java
##########
@@ -0,0 +1,137 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.scheduler;
+
+import org.apache.flink.runtime.concurrent.FutureUtils;
+import org.apache.flink.runtime.executiongraph.ExecutionVertex;
+import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
+import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+import static
org.apache.flink.runtime.executiongraph.ExecutionVertex.MAX_DISTINCT_LOCATIONS_TO_CONSIDER;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Default implementation of {@link PreferredLocationsRetriever}.
+ * Locations based on state will be returned if exist.
+ * Otherwise locations based on inputs will be returned.
+ */
+public class DefaultPreferredLocationsRetriever implements
PreferredLocationsRetriever {
+
+ private final StateLocationRetriever stateLocationRetriever;
+
+ private final InputsLocationsRetriever inputsLocationsRetriever;
+
+ DefaultPreferredLocationsRetriever(
+ final StateLocationRetriever stateLocationRetriever,
+ final InputsLocationsRetriever
inputsLocationsRetriever) {
+
+ this.stateLocationRetriever =
checkNotNull(stateLocationRetriever);
+ this.inputsLocationsRetriever =
checkNotNull(inputsLocationsRetriever);
+ }
+
+ @Override
+ public CompletableFuture<Collection<TaskManagerLocation>>
getPreferredLocations(
+ final ExecutionVertexID executionVertexId,
+ final Set<ExecutionVertexID> producersToIgnore) {
+
+ checkNotNull(executionVertexId);
+ checkNotNull(producersToIgnore);
+
+ final Collection<TaskManagerLocation>
preferredLocationsBasedOnState =
+ getPreferredLocationsBasedOnState(executionVertexId,
stateLocationRetriever);
+ if (!preferredLocationsBasedOnState.isEmpty()) {
+ return
CompletableFuture.completedFuture(preferredLocationsBasedOnState);
+ }
+
+ return getPreferredLocationsBasedOnInputs(executionVertexId,
producersToIgnore, inputsLocationsRetriever);
+ }
+
+ private static Collection<TaskManagerLocation>
getPreferredLocationsBasedOnState(
+ final ExecutionVertexID executionVertexId,
+ final StateLocationRetriever stateLocationRetriever) {
+
+ return
stateLocationRetriever.getStateLocation(executionVertexId)
+ .map(Collections::singleton)
+ .orElse(Collections.emptySet());
+ }
+
+ /**
+ * Gets the location preferences of the execution, as determined by the
locations
+ * of the predecessors from which it receives input data.
+ * If there are more than {@link
ExecutionVertex#MAX_DISTINCT_LOCATIONS_TO_CONSIDER} different locations of
source data,
+ * or neither the sources have not been started nor will be started
with the execution together,
+ * this method returns an empty collection to indicate no location
preference.
+ *
+ * @return The preferred locations based in input streams, or an empty
iterable,
+ * if there is no input-based preference.
+ */
+ static CompletableFuture<Collection<TaskManagerLocation>>
getPreferredLocationsBasedOnInputs(
+ final ExecutionVertexID executionVertexId,
+ final Set<ExecutionVertexID> producersToIgnore,
+ final InputsLocationsRetriever
inputsLocationsRetriever) {
+
+ CompletableFuture<Collection<TaskManagerLocation>>
preferredLocations =
+
CompletableFuture.completedFuture(Collections.emptyList());
+
+ final Collection<CompletableFuture<TaskManagerLocation>>
locationsFutures = new ArrayList<>();
+
+ final Collection<Collection<ExecutionVertexID>> allProducers =
+
inputsLocationsRetriever.getConsumedResultPartitionsProducers(executionVertexId);
+ for (Collection<ExecutionVertexID> producers : allProducers) {
+
+ for (ExecutionVertexID producer : producers) {
+ final
Optional<CompletableFuture<TaskManagerLocation>> optionalLocationFuture;
+ if (!producersToIgnore.contains(producer)) {
+ optionalLocationFuture =
inputsLocationsRetriever.getTaskManagerLocation(producer);
+ } else {
+ optionalLocationFuture =
Optional.empty();
+ }
+
optionalLocationFuture.ifPresent(locationsFutures::add);
+ // If the parallelism is large, wait for all
futures coming back may cost a long time.
+ if (locationsFutures.size() >
MAX_DISTINCT_LOCATIONS_TO_CONSIDER) {
+ locationsFutures.clear();
+ break;
+ }
+ }
Review comment:
Could this be a separate method?
```
private Collection<CompletableFuture<TaskManagerLocation>>
getInputLocationFutures(
Set<ExecutionVertexID> producersToIgnore,
Collection<ExecutionVertexID> producers)
```
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultPreferredLocationsRetriever.java
##########
@@ -0,0 +1,137 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.scheduler;
+
+import org.apache.flink.runtime.concurrent.FutureUtils;
+import org.apache.flink.runtime.executiongraph.ExecutionVertex;
+import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
+import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+
+import static
org.apache.flink.runtime.executiongraph.ExecutionVertex.MAX_DISTINCT_LOCATIONS_TO_CONSIDER;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Default implementation of {@link PreferredLocationsRetriever}.
+ * Locations based on state will be returned if exist.
+ * Otherwise locations based on inputs will be returned.
+ */
+public class DefaultPreferredLocationsRetriever implements
PreferredLocationsRetriever {
+
+ private final StateLocationRetriever stateLocationRetriever;
+
+ private final InputsLocationsRetriever inputsLocationsRetriever;
+
+ DefaultPreferredLocationsRetriever(
+ final StateLocationRetriever stateLocationRetriever,
+ final InputsLocationsRetriever
inputsLocationsRetriever) {
+
+ this.stateLocationRetriever =
checkNotNull(stateLocationRetriever);
+ this.inputsLocationsRetriever =
checkNotNull(inputsLocationsRetriever);
+ }
+
+ @Override
+ public CompletableFuture<Collection<TaskManagerLocation>>
getPreferredLocations(
+ final ExecutionVertexID executionVertexId,
+ final Set<ExecutionVertexID> producersToIgnore) {
+
+ checkNotNull(executionVertexId);
+ checkNotNull(producersToIgnore);
+
+ final Collection<TaskManagerLocation>
preferredLocationsBasedOnState =
+ getPreferredLocationsBasedOnState(executionVertexId,
stateLocationRetriever);
+ if (!preferredLocationsBasedOnState.isEmpty()) {
+ return
CompletableFuture.completedFuture(preferredLocationsBasedOnState);
+ }
+
+ return getPreferredLocationsBasedOnInputs(executionVertexId,
producersToIgnore, inputsLocationsRetriever);
+ }
+
+ private static Collection<TaskManagerLocation>
getPreferredLocationsBasedOnState(
+ final ExecutionVertexID executionVertexId,
+ final StateLocationRetriever stateLocationRetriever) {
+
+ return
stateLocationRetriever.getStateLocation(executionVertexId)
+ .map(Collections::singleton)
+ .orElse(Collections.emptySet());
+ }
+
+ /**
+ * Gets the location preferences of the execution, as determined by the
locations
+ * of the predecessors from which it receives input data.
+ * If there are more than {@link
ExecutionVertex#MAX_DISTINCT_LOCATIONS_TO_CONSIDER} different locations of
source data,
+ * or neither the sources have not been started nor will be started
with the execution together,
+ * this method returns an empty collection to indicate no location
preference.
+ *
+ * @return The preferred locations based in input streams, or an empty
iterable,
+ * if there is no input-based preference.
+ */
+ static CompletableFuture<Collection<TaskManagerLocation>>
getPreferredLocationsBasedOnInputs(
+ final ExecutionVertexID executionVertexId,
+ final Set<ExecutionVertexID> producersToIgnore,
Review comment:
```suggestion
final Collection<ExecutionVertexID> producersToIgnore,
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
