You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2022/03/21 11:49:14 UTC
[flink] branch master updated: Partially revert "[hotfix][runtime] Remove unnecessary method."
This is an automated email from the ASF dual-hosted git repository.
pnowojski 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 a6b586a Partially revert "[hotfix][runtime] Remove unnecessary method."
a6b586a is described below
commit a6b586aaa49e10a9793b58fee5e92ba5bc098959
Author: Piotr Nowojski <pi...@gmail.com>
AuthorDate: Mon Mar 21 12:47:14 2022 +0100
Partially revert "[hotfix][runtime] Remove unnecessary method."
This reverts commit a1473daaffbab7cf7abc974fadb634408eeb3db7.
---
.../scheduler/strategy/SchedulingStrategyUtils.java | 17 +++++++++++++++++
1 file changed, 17 insertions(+)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/strategy/SchedulingStrategyUtils.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/strategy/SchedulingStrategyUtils.java
index 4af58d2..efa2cb4 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/strategy/SchedulingStrategyUtils.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/strategy/SchedulingStrategyUtils.java
@@ -33,6 +33,23 @@ import java.util.stream.Collectors;
/** Utils for {@link SchedulingStrategy}. */
class SchedulingStrategyUtils {
+ static List<ExecutionVertexDeploymentOption>
+ createExecutionVertexDeploymentOptionsInTopologicalOrder(
+ final SchedulingTopology topology,
+ final Set<ExecutionVertexID> verticesToDeploy,
+ final Function<ExecutionVertexID, DeploymentOption> deploymentOptionRetriever) {
+
+ return IterableUtils.toStream(topology.getVertices())
+ .map(SchedulingExecutionVertex::getId)
+ .filter(verticesToDeploy::contains)
+ .map(
+ executionVertexID ->
+ new ExecutionVertexDeploymentOption(
+ executionVertexID,
+ deploymentOptionRetriever.apply(executionVertexID)))
+ .collect(Collectors.toList());
+ }
+
static List<ExecutionVertexDeploymentOption> createExecutionVertexDeploymentOptions(
final Collection<ExecutionVertexID> verticesToDeploy,
final Function<ExecutionVertexID, DeploymentOption> deploymentOptionRetriever) {