You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ro...@apache.org on 2023/03/02 21:39:28 UTC
[flink] branch master updated (18c73f1488b -> e38a6709b57)
This is an automated email from the ASF dual-hosted git repository.
roman pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from 18c73f1488b [docs][hotfix] correct label in config configuration
new d718342ad03 [hotfix][tests] Make LocalRecoveryITCase fail when allocations don't match
new 676e8e5528f [FLINK-21450][runtime] Add previous ExecutionGraph to WaitingForResources AdaptiveScheduler state
new 416cb7aaa02 [FLINK-21450][runtime] Restructure types passed between AdaptiveScheduler and SlotAssigner
new e38a6709b57 [FLINK-21450][runtime] Support LocalRecovery by AdaptiveScheduler
The 4 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../jobgraph/jsonplan/JsonPlanGenerator.java | 21 +-
.../scheduler/adaptive/AdaptiveScheduler.java | 50 ++---
.../flink/runtime/scheduler/adaptive/Created.java | 2 +-
.../scheduler/adaptive/CreatingExecutionGraph.java | 31 ++-
.../scheduler/adaptive/JobGraphJobInformation.java | 5 +
.../scheduler/adaptive/JobSchedulingPlan.java | 105 ++++++++++
.../runtime/scheduler/adaptive/Restarting.java | 5 +-
.../scheduler/adaptive/StateTransitions.java | 6 +-
.../scheduler/adaptive/WaitingForResources.java | 28 ++-
.../adaptive/allocator/DefaultSlotAssigner.java | 77 ++++++++
.../allocator/JobAllocationsInformation.java | 112 +++++++++++
.../adaptive/allocator/JobInformation.java | 2 +
.../adaptive/allocator/SlotAllocator.java | 18 +-
.../{VertexParallelism.java => SlotAssigner.java} | 23 ++-
.../allocator/SlotSharingSlotAllocator.java | 123 +++++-------
.../allocator/StateLocalitySlotAssigner.java | 201 +++++++++++++++++++
.../adaptive/allocator/VertexParallelism.java | 43 +++-
.../VertexParallelismWithSlotSharing.java | 52 -----
.../scheduler/adaptive/AdaptiveSchedulerTest.java | 3 +-
.../runtime/scheduler/adaptive/CreatedTest.java | 3 +-
.../adaptive/CreatingExecutionGraphTest.java | 52 +++--
.../runtime/scheduler/adaptive/RestartingTest.java | 2 +-
.../adaptive/WaitingForResourcesTest.java | 9 +-
.../allocator/SlotSharingSlotAllocatorTest.java | 219 +++++++++++----------
.../allocator/StateLocalitySlotAssignerTest.java | 166 ++++++++++++++++
.../adaptive/allocator/TestJobInformation.java | 59 ++++++
.../scheduler/adaptive/allocator/TestSlotInfo.java | 10 +-
.../adaptive/allocator/TestVertexInformation.java} | 49 +++--
.../adaptive/allocator/TestingSlotAllocator.java | 49 ++---
.../flink/test/recovery/LocalRecoveryITCase.java | 28 ++-
30 files changed, 1135 insertions(+), 418 deletions(-)
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/JobSchedulingPlan.java
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/DefaultSlotAssigner.java
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/JobAllocationsInformation.java
copy flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/{VertexParallelism.java => SlotAssigner.java} (62%)
create mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/StateLocalitySlotAssigner.java
delete mode 100644 flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/VertexParallelismWithSlotSharing.java
create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/StateLocalitySlotAssignerTest.java
create mode 100644 flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/TestJobInformation.java
copy flink-runtime/src/{main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/JobInformation.java => test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/TestVertexInformation.java} (53%)
[flink] 04/04: [FLINK-21450][runtime] Support LocalRecovery by AdaptiveScheduler
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
roman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit e38a6709b57000e38bf044d0e55da3dd3ec3bde8
Author: David Moravek <dm...@apache.org>
AuthorDate: Fri Jan 21 10:09:53 2022 +0100
[FLINK-21450][runtime] Support LocalRecovery by AdaptiveScheduler
---
.../scheduler/adaptive/AdaptiveScheduler.java | 16 +-
.../scheduler/adaptive/JobGraphJobInformation.java | 5 +
.../adaptive/allocator/DefaultSlotAssigner.java | 77 ++++++++
.../allocator/JobAllocationsInformation.java | 112 ++++++++++++
.../adaptive/allocator/JobInformation.java | 2 +
.../adaptive/allocator/SlotAllocator.java | 4 +-
.../{JobInformation.java => SlotAssigner.java} | 36 ++--
.../allocator/SlotSharingSlotAllocator.java | 70 +++----
.../allocator/StateLocalitySlotAssigner.java | 201 +++++++++++++++++++++
.../allocator/SlotSharingSlotAllocatorTest.java | 161 +++++++++--------
.../allocator/StateLocalitySlotAssignerTest.java | 166 +++++++++++++++++
.../adaptive/allocator/TestJobInformation.java | 59 ++++++
.../scheduler/adaptive/allocator/TestSlotInfo.java | 10 +-
.../adaptive/allocator/TestVertexInformation.java} | 49 ++---
.../adaptive/allocator/TestingSlotAllocator.java | 31 +---
15 files changed, 802 insertions(+), 197 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
index a876bb65eb2..2446d41c5c9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
@@ -96,6 +96,7 @@ import org.apache.flink.runtime.scheduler.SchedulerUtils;
import org.apache.flink.runtime.scheduler.UpdateSchedulerNgOnInternalFailuresListener;
import org.apache.flink.runtime.scheduler.VertexParallelismInformation;
import org.apache.flink.runtime.scheduler.VertexParallelismStore;
+import org.apache.flink.runtime.scheduler.adaptive.allocator.JobAllocationsInformation;
import org.apache.flink.runtime.scheduler.adaptive.allocator.ReservedSlots;
import org.apache.flink.runtime.scheduler.adaptive.allocator.SlotAllocator;
import org.apache.flink.runtime.scheduler.adaptive.allocator.VertexParallelism;
@@ -760,12 +761,15 @@ public class AdaptiveScheduler
.isPresent();
}
- private JobSchedulingPlan determineParallelism(SlotAllocator slotAllocator)
+ private JobSchedulingPlan determineParallelism(
+ SlotAllocator slotAllocator, @Nullable ExecutionGraph previousExecutionGraph)
throws NoResourceAvailableException {
return slotAllocator
.determineParallelismAndCalculateAssignment(
- jobInformation, declarativeSlotPool.getFreeSlotsInformation())
+ jobInformation,
+ declarativeSlotPool.getFreeSlotsInformation(),
+ JobAllocationsInformation.fromGraph(previousExecutionGraph))
.orElseThrow(
() ->
new NoResourceAvailableException(
@@ -921,8 +925,7 @@ public class AdaptiveScheduler
public void goToCreatingExecutionGraph(@Nullable ExecutionGraph previousExecutionGraph) {
final CompletableFuture<CreatingExecutionGraph.ExecutionGraphWithVertexParallelism>
executionGraphWithAvailableResourcesFuture =
- createExecutionGraphWithAvailableResourcesAsync();
-
+ createExecutionGraphWithAvailableResourcesAsync(previousExecutionGraph);
transitionToState(
new CreatingExecutionGraph.Factory(
this,
@@ -932,12 +935,13 @@ public class AdaptiveScheduler
}
private CompletableFuture<CreatingExecutionGraph.ExecutionGraphWithVertexParallelism>
- createExecutionGraphWithAvailableResourcesAsync() {
+ createExecutionGraphWithAvailableResourcesAsync(
+ @Nullable ExecutionGraph previousExecutionGraph) {
final JobSchedulingPlan schedulingPlan;
final VertexParallelismStore adjustedParallelismStore;
try {
- schedulingPlan = determineParallelism(slotAllocator);
+ schedulingPlan = determineParallelism(slotAllocator, previousExecutionGraph);
JobGraph adjustedJobGraph = jobInformation.copyJobGraph();
for (JobVertex vertex : adjustedJobGraph.getVertices()) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/JobGraphJobInformation.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/JobGraphJobInformation.java
index 1b0bb60e7e6..2111b6e9c07 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/JobGraphJobInformation.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/JobGraphJobInformation.java
@@ -106,6 +106,11 @@ public class JobGraphJobInformation implements JobInformation {
return parallelismInfo.getParallelism();
}
+ @Override
+ public int getMaxParallelism() {
+ return parallelismInfo.getMaxParallelism();
+ }
+
@Override
public SlotSharingGroup getSlotSharingGroup() {
return jobVertex.getSlotSharingGroup();
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/DefaultSlotAssigner.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/DefaultSlotAssigner.java
new file mode 100644
index 00000000000..0a8813b33a6
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/DefaultSlotAssigner.java
@@ -0,0 +1,77 @@
+/*
+ * 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.adaptive.allocator;
+
+import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
+import org.apache.flink.runtime.jobmaster.SlotInfo;
+import org.apache.flink.runtime.scheduler.adaptive.JobSchedulingPlan.SlotAssignment;
+import org.apache.flink.runtime.scheduler.adaptive.allocator.SlotSharingSlotAllocator.ExecutionSlotSharingGroup;
+import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/** Simple {@link SlotAssigner} that treats all slots and slot sharing groups equally. */
+public class DefaultSlotAssigner implements SlotAssigner {
+
+ @Override
+ public Collection<SlotAssignment> assignSlots(
+ JobInformation jobInformation,
+ Collection<? extends SlotInfo> freeSlots,
+ VertexParallelism vertexParallelism,
+ JobAllocationsInformation previousAllocations) {
+ List<ExecutionSlotSharingGroup> allGroups = new ArrayList<>();
+ for (SlotSharingGroup slotSharingGroup : jobInformation.getSlotSharingGroups()) {
+ allGroups.addAll(createExecutionSlotSharingGroups(vertexParallelism, slotSharingGroup));
+ }
+
+ Iterator<? extends SlotInfo> iterator = freeSlots.iterator();
+ Collection<SlotAssignment> assignments = new ArrayList<>();
+ for (ExecutionSlotSharingGroup group : allGroups) {
+ assignments.add(new SlotAssignment(iterator.next(), group));
+ }
+ return assignments;
+ }
+
+ static List<ExecutionSlotSharingGroup> createExecutionSlotSharingGroups(
+ VertexParallelism vertexParallelism, SlotSharingGroup slotSharingGroup) {
+ final Map<Integer, Set<ExecutionVertexID>> sharedSlotToVertexAssignment = new HashMap<>();
+ slotSharingGroup
+ .getJobVertexIds()
+ .forEach(
+ jobVertexId -> {
+ int parallelism = vertexParallelism.getParallelism(jobVertexId);
+ for (int subtaskIdx = 0; subtaskIdx < parallelism; subtaskIdx++) {
+ sharedSlotToVertexAssignment
+ .computeIfAbsent(subtaskIdx, ignored -> new HashSet<>())
+ .add(new ExecutionVertexID(jobVertexId, subtaskIdx));
+ }
+ });
+ return sharedSlotToVertexAssignment.values().stream()
+ .map(ExecutionSlotSharingGroup::new)
+ .collect(Collectors.toList());
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/JobAllocationsInformation.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/JobAllocationsInformation.java
new file mode 100644
index 00000000000..e17d242d342
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/JobAllocationsInformation.java
@@ -0,0 +1,112 @@
+/*
+ * 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.adaptive.allocator;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+import org.apache.flink.runtime.executiongraph.ExecutionGraph;
+import org.apache.flink.runtime.executiongraph.ExecutionJobVertex;
+import org.apache.flink.runtime.executiongraph.ExecutionVertex;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyGroupRangeAssignment;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static java.util.Collections.emptyList;
+import static java.util.Collections.emptyMap;
+
+/** Information about allocations of Job Vertices. */
+@Internal
+public class JobAllocationsInformation {
+
+ private final Map<JobVertexID, List<VertexAllocationInformation>> vertexAllocations;
+
+ JobAllocationsInformation(
+ Map<JobVertexID, List<VertexAllocationInformation>> vertexAllocations) {
+ this.vertexAllocations = vertexAllocations;
+ }
+
+ public static JobAllocationsInformation fromGraph(@Nullable ExecutionGraph graph) {
+ return graph == null ? empty() : new JobAllocationsInformation(calculateAllocations(graph));
+ }
+
+ public List<VertexAllocationInformation> getAllocations(JobVertexID jobVertexID) {
+ return vertexAllocations.getOrDefault(jobVertexID, emptyList());
+ }
+
+ private static Map<JobVertexID, List<VertexAllocationInformation>> calculateAllocations(
+ ExecutionGraph graph) {
+ final Map<JobVertexID, List<VertexAllocationInformation>> allocations = new HashMap<>();
+ for (ExecutionJobVertex vertex : graph.getVerticesTopologically()) {
+ JobVertexID jobVertexId = vertex.getJobVertexId();
+ for (ExecutionVertex executionVertex : vertex.getTaskVertices()) {
+ AllocationID allocationId =
+ executionVertex.getCurrentExecutionAttempt().getAssignedAllocationID();
+ KeyGroupRange kgr =
+ KeyGroupRangeAssignment.computeKeyGroupRangeForOperatorIndex(
+ vertex.getMaxParallelism(),
+ vertex.getParallelism(),
+ executionVertex.getParallelSubtaskIndex());
+ allocations
+ .computeIfAbsent(jobVertexId, ignored -> new ArrayList<>())
+ .add(new VertexAllocationInformation(allocationId, jobVertexId, kgr));
+ }
+ }
+ return allocations;
+ }
+
+ public static JobAllocationsInformation empty() {
+ return new JobAllocationsInformation(emptyMap());
+ }
+
+ public boolean isEmpty() {
+ return vertexAllocations.isEmpty();
+ }
+
+ /** Information about the allocations of a single Job Vertex. */
+ public static class VertexAllocationInformation {
+ private final AllocationID allocationID;
+ private final JobVertexID jobVertexID;
+ private final KeyGroupRange keyGroupRange;
+
+ public VertexAllocationInformation(
+ AllocationID allocationID, JobVertexID jobVertexID, KeyGroupRange keyGroupRange) {
+ this.allocationID = allocationID;
+ this.jobVertexID = jobVertexID;
+ this.keyGroupRange = keyGroupRange;
+ }
+
+ public AllocationID getAllocationID() {
+ return allocationID;
+ }
+
+ public JobVertexID getJobVertexID() {
+ return jobVertexID;
+ }
+
+ public KeyGroupRange getKeyGroupRange() {
+ return keyGroupRange;
+ }
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/JobInformation.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/JobInformation.java
index 99b8d48c051..b1ff9183b6a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/JobInformation.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/JobInformation.java
@@ -44,6 +44,8 @@ public interface JobInformation {
int getParallelism();
+ int getMaxParallelism();
+
SlotSharingGroup getSlotSharingGroup();
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotAllocator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotAllocator.java
index fea0d659ad3..629a57f5122 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotAllocator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotAllocator.java
@@ -60,7 +60,9 @@ public interface SlotAllocator {
* assignment of slots to execution slot sharing groups.
*/
Optional<JobSchedulingPlan> determineParallelismAndCalculateAssignment(
- JobInformation jobInformation, Collection<? extends SlotInfo> slots);
+ JobInformation jobInformation,
+ Collection<? extends SlotInfo> slots,
+ JobAllocationsInformation jobAllocationsInformation);
/**
* Reserves slots according to the given assignment if possible. If the underlying set of
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/JobInformation.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotAssigner.java
similarity index 52%
copy from flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/JobInformation.java
copy to flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotAssigner.java
index 99b8d48c051..cb264e28d88 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/JobInformation.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotAssigner.java
@@ -17,33 +17,19 @@
package org.apache.flink.runtime.scheduler.adaptive.allocator;
-import org.apache.flink.runtime.jobgraph.JobVertexID;
-import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.jobmaster.SlotInfo;
+import org.apache.flink.runtime.scheduler.adaptive.JobSchedulingPlan.SlotAssignment;
import java.util.Collection;
-/** Information about the job. */
-public interface JobInformation {
- /**
- * Returns all slot-sharing groups of the job.
- *
- * <p>Attention: The returned slot sharing groups should never be modified (they are indeed
- * mutable)!
- *
- * @return all slot-sharing groups of the job
- */
- Collection<SlotSharingGroup> getSlotSharingGroups();
+/** Interface for assigning slots to slot sharing groups. */
+@Internal
+public interface SlotAssigner {
- VertexInformation getVertexInformation(JobVertexID jobVertexId);
-
- Iterable<VertexInformation> getVertices();
-
- /** Information about a single vertex. */
- interface VertexInformation {
- JobVertexID getJobVertexID();
-
- int getParallelism();
-
- SlotSharingGroup getSlotSharingGroup();
- }
+ Collection<SlotAssignment> assignSlots(
+ JobInformation jobInformation,
+ Collection<? extends SlotInfo> freeSlots,
+ VertexParallelism vertexParallelism,
+ JobAllocationsInformation previousAllocations);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotSharingSlotAllocator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotSharingSlotAllocator.java
index 2b7a9c362ad..3d0f30b69e2 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotSharingSlotAllocator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotSharingSlotAllocator.java
@@ -37,12 +37,11 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.Comparator;
import java.util.HashMap;
-import java.util.HashSet;
-import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.UUID;
import java.util.stream.Collectors;
/** {@link SlotAllocator} implementation that supports slot sharing. */
@@ -125,30 +124,24 @@ public class SlotSharingSlotAllocator implements SlotAllocator {
@Override
public Optional<JobSchedulingPlan> determineParallelismAndCalculateAssignment(
- JobInformation jobInformation, Collection<? extends SlotInfo> slots) {
+ JobInformation jobInformation,
+ Collection<? extends SlotInfo> slots,
+ JobAllocationsInformation jobAllocationsInformation) {
return determineParallelism(jobInformation, slots)
.map(
- parallelism ->
- new JobSchedulingPlan(
- parallelism,
- assignSlots(jobInformation, slots, parallelism)));
- }
-
- private Collection<SlotAssignment> assignSlots(
- JobInformation jobInformation,
- Collection<? extends SlotInfo> freeSlots,
- VertexParallelism vertexParallelism) {
- List<ExecutionSlotSharingGroup> allGroups = new ArrayList<>();
- for (SlotSharingGroup slotSharingGroup : jobInformation.getSlotSharingGroups()) {
- allGroups.addAll(createExecutionSlotSharingGroups(vertexParallelism, slotSharingGroup));
- }
-
- Iterator<? extends SlotInfo> iterator = freeSlots.iterator();
- Collection<SlotAssignment> assignments = new ArrayList<>();
- for (ExecutionSlotSharingGroup group : allGroups) {
- assignments.add(new SlotAssignment(iterator.next(), group));
- }
- return assignments;
+ parallelism -> {
+ SlotAssigner slotAssigner =
+ jobAllocationsInformation.isEmpty()
+ ? new DefaultSlotAssigner()
+ : new StateLocalitySlotAssigner();
+ return new JobSchedulingPlan(
+ parallelism,
+ slotAssigner.assignSlots(
+ jobInformation,
+ slots,
+ parallelism,
+ jobAllocationsInformation));
+ });
}
/**
@@ -200,24 +193,6 @@ public class SlotSharingSlotAllocator implements SlotAllocator {
return vertexParallelism;
}
- private static List<ExecutionSlotSharingGroup> createExecutionSlotSharingGroups(
- VertexParallelism vertexParallelism, SlotSharingGroup slotSharingGroup) {
- final Map<Integer, Set<ExecutionVertexID>> sharedSlotToVertexAssignment = new HashMap<>();
-
- for (JobVertexID jobVertexId : slotSharingGroup.getJobVertexIds()) {
- int parallelism = vertexParallelism.getParallelism(jobVertexId);
- for (int subtaskIdx = 0; subtaskIdx < parallelism; subtaskIdx++) {
- sharedSlotToVertexAssignment
- .computeIfAbsent(subtaskIdx, ignored -> new HashSet<>())
- .add(new ExecutionVertexID(jobVertexId, subtaskIdx));
- }
- }
-
- return sharedSlotToVertexAssignment.values().stream()
- .map(ExecutionSlotSharingGroup::new)
- .collect(Collectors.toList());
- }
-
@Override
public Optional<ReservedSlots> tryReserveResources(JobSchedulingPlan jobSchedulingPlan) {
final Collection<AllocationID> expectedSlots =
@@ -278,10 +253,21 @@ public class SlotSharingSlotAllocator implements SlotAllocator {
}
static class ExecutionSlotSharingGroup {
+ private final String id;
private final Set<ExecutionVertexID> containedExecutionVertices;
public ExecutionSlotSharingGroup(Set<ExecutionVertexID> containedExecutionVertices) {
+ this(containedExecutionVertices, UUID.randomUUID().toString());
+ }
+
+ public ExecutionSlotSharingGroup(
+ Set<ExecutionVertexID> containedExecutionVertices, String id) {
this.containedExecutionVertices = containedExecutionVertices;
+ this.id = id;
+ }
+
+ public String getId() {
+ return id;
}
public Collection<ExecutionVertexID> getContainedExecutionVertices() {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/StateLocalitySlotAssigner.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/StateLocalitySlotAssigner.java
new file mode 100644
index 00000000000..4e2a807c062
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/StateLocalitySlotAssigner.java
@@ -0,0 +1,201 @@
+/*
+ * 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.adaptive.allocator;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
+import org.apache.flink.runtime.jobmaster.SlotInfo;
+import org.apache.flink.runtime.scheduler.adaptive.JobSchedulingPlan.SlotAssignment;
+import org.apache.flink.runtime.scheduler.adaptive.allocator.SlotSharingSlotAllocator.ExecutionSlotSharingGroup;
+import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyGroupRangeAssignment;
+
+import javax.annotation.Nonnull;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.PriorityQueue;
+import java.util.stream.Collectors;
+
+import static java.util.function.Function.identity;
+import static java.util.stream.Collectors.toMap;
+import static org.apache.flink.runtime.scheduler.adaptive.allocator.DefaultSlotAssigner.createExecutionSlotSharingGroups;
+import static org.apache.flink.util.Preconditions.checkState;
+
+/** A {@link SlotAssigner} that assigns slots based on the number of local key groups. */
+@Internal
+public class StateLocalitySlotAssigner implements SlotAssigner {
+
+ private static class AllocationScore implements Comparable<AllocationScore> {
+
+ private final String groupId;
+ private final AllocationID allocationId;
+
+ public AllocationScore(String groupId, AllocationID allocationId, long score) {
+ this.groupId = groupId;
+ this.allocationId = allocationId;
+ this.score = score;
+ }
+
+ private final long score;
+
+ public String getGroupId() {
+ return groupId;
+ }
+
+ public AllocationID getAllocationId() {
+ return allocationId;
+ }
+
+ public long getScore() {
+ return score;
+ }
+
+ @Override
+ public int compareTo(StateLocalitySlotAssigner.AllocationScore other) {
+ int result = Long.compare(score, other.score);
+ if (result != 0) {
+ return result;
+ }
+ result = other.allocationId.compareTo(allocationId);
+ if (result != 0) {
+ return result;
+ }
+ return other.groupId.compareTo(groupId);
+ }
+ }
+
+ @Override
+ public Collection<SlotAssignment> assignSlots(
+ JobInformation jobInformation,
+ Collection<? extends SlotInfo> freeSlots,
+ VertexParallelism vertexParallelism,
+ JobAllocationsInformation previousAllocations) {
+ checkState(
+ freeSlots.size() >= jobInformation.getSlotSharingGroups().size(),
+ "Not enough slots to allocate all the slot sharing groups (have: %s, need: %s)",
+ freeSlots.size(),
+ jobInformation.getSlotSharingGroups().size());
+
+ final List<ExecutionSlotSharingGroup> allGroups = new ArrayList<>();
+ for (SlotSharingGroup slotSharingGroup : jobInformation.getSlotSharingGroups()) {
+ allGroups.addAll(createExecutionSlotSharingGroups(vertexParallelism, slotSharingGroup));
+ }
+ final Map<JobVertexID, Integer> parallelism = getParallelism(allGroups);
+ final PriorityQueue<AllocationScore> scores =
+ calculateScores(jobInformation, previousAllocations, allGroups, parallelism);
+
+ final Map<String, ExecutionSlotSharingGroup> groupsById =
+ allGroups.stream().collect(toMap(ExecutionSlotSharingGroup::getId, identity()));
+ final Map<AllocationID, SlotInfo> slotsById =
+ freeSlots.stream().collect(toMap(SlotInfo::getAllocationId, identity()));
+ AllocationScore score;
+ final Collection<SlotAssignment> assignments = new ArrayList<>();
+ while ((score = scores.poll()) != null) {
+ if (slotsById.containsKey(score.getAllocationId())
+ && groupsById.containsKey(score.getGroupId())) {
+ assignments.add(
+ new SlotAssignment(
+ slotsById.remove(score.getAllocationId()),
+ groupsById.remove(score.getGroupId())));
+ }
+ }
+ // Distribute the remaining slots with no score
+ Iterator<? extends SlotInfo> remainingSlots = slotsById.values().iterator();
+ for (ExecutionSlotSharingGroup group : groupsById.values()) {
+ checkState(
+ remainingSlots.hasNext(),
+ "No slots available for group %s (%s more in total). This is likely a bug.",
+ group,
+ groupsById.size());
+ assignments.add(new SlotAssignment(remainingSlots.next(), group));
+ remainingSlots.remove();
+ }
+
+ return assignments;
+ }
+
+ @Nonnull
+ private PriorityQueue<AllocationScore> calculateScores(
+ JobInformation jobInformation,
+ JobAllocationsInformation previousAllocations,
+ List<ExecutionSlotSharingGroup> allGroups,
+ Map<JobVertexID, Integer> parallelism) {
+ // PQ orders the pairs (allocationID, groupID) by score, decreasing
+ // the score is computed as the potential amount of state that would reside locally
+ final PriorityQueue<AllocationScore> scores =
+ new PriorityQueue<>(Comparator.reverseOrder());
+ for (ExecutionSlotSharingGroup group : allGroups) {
+ scores.addAll(calculateScore(group, parallelism, jobInformation, previousAllocations));
+ }
+ return scores;
+ }
+
+ private static Map<JobVertexID, Integer> getParallelism(
+ List<ExecutionSlotSharingGroup> groups) {
+ final Map<JobVertexID, Integer> parallelism = new HashMap<>();
+ for (ExecutionSlotSharingGroup group : groups) {
+ for (ExecutionVertexID evi : group.getContainedExecutionVertices()) {
+ parallelism.merge(evi.getJobVertexId(), 1, Integer::sum);
+ }
+ }
+ return parallelism;
+ }
+
+ public Collection<AllocationScore> calculateScore(
+ ExecutionSlotSharingGroup group,
+ Map<JobVertexID, Integer> parallelism,
+ JobInformation jobInformation,
+ JobAllocationsInformation previousAllocations) {
+ final Map<AllocationID, Long> score = new HashMap<>();
+ for (ExecutionVertexID evi : group.getContainedExecutionVertices()) {
+ final KeyGroupRange kgr =
+ KeyGroupRangeAssignment.computeKeyGroupRangeForOperatorIndex(
+ jobInformation
+ .getVertexInformation(evi.getJobVertexId())
+ .getMaxParallelism(),
+ parallelism.get(evi.getJobVertexId()),
+ evi.getSubtaskIndex());
+ previousAllocations
+ .getAllocations(evi.getJobVertexId())
+ .forEach(
+ allocation -> {
+ long value =
+ allocation
+ .getKeyGroupRange()
+ .getIntersection(kgr)
+ .getNumberOfKeyGroups();
+ if (value > 0) {
+ score.merge(allocation.getAllocationID(), value, Long::sum);
+ }
+ });
+ }
+
+ return score.entrySet().stream()
+ .map(e -> new AllocationScore(group.getId(), e.getKey(), e.getValue()))
+ .collect(Collectors.toList());
+ }
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotSharingSlotAllocatorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotSharingSlotAllocatorTest.java
index 7643acc7410..9a6b1cab5c2 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotSharingSlotAllocatorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotSharingSlotAllocatorTest.java
@@ -17,6 +17,7 @@
package org.apache.flink.runtime.scheduler.adaptive.allocator;
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
@@ -24,25 +25,31 @@ import org.apache.flink.runtime.jobmaster.LogicalSlot;
import org.apache.flink.runtime.jobmaster.SlotInfo;
import org.apache.flink.runtime.scheduler.TestingPhysicalSlot;
import org.apache.flink.runtime.scheduler.adaptive.JobSchedulingPlan;
+import org.apache.flink.runtime.scheduler.adaptive.allocator.JobAllocationsInformation.VertexAllocationInformation;
import org.apache.flink.runtime.scheduler.adaptive.allocator.SlotSharingSlotAllocator.ExecutionSlotSharingGroup;
import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.topology.VertexID;
import org.apache.flink.runtime.util.ResourceCounter;
import org.apache.flink.util.TestLogger;
import org.assertj.core.api.Assertions;
-import org.junit.Test;
+import org.junit.jupiter.api.Test;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
import java.util.Map;
import java.util.Optional;
-import java.util.function.Function;
-import java.util.stream.Collectors;
+import java.util.Set;
+import java.util.stream.IntStream;
+import static org.assertj.core.api.Assertions.assertThat;
import static org.hamcrest.CoreMatchers.is;
-import static org.hamcrest.Matchers.contains;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertThat;
@@ -79,12 +86,11 @@ public class SlotSharingSlotAllocatorTest extends TestLogger {
final ResourceCounter resourceCounter =
slotAllocator.calculateRequiredSlots(Arrays.asList(vertex1, vertex2, vertex3));
- assertThat(resourceCounter.getResources(), contains(ResourceProfile.UNKNOWN));
- assertThat(
- resourceCounter.getResourceCount(ResourceProfile.UNKNOWN),
- is(
+ assertThat(resourceCounter.getResources()).contains(ResourceProfile.UNKNOWN);
+ assertThat(resourceCounter.getResourceCount(ResourceProfile.UNKNOWN))
+ .isEqualTo(
Math.max(vertex1.getParallelism(), vertex2.getParallelism())
- + vertex3.getParallelism()));
+ + vertex3.getParallelism());
}
@Test
@@ -194,7 +200,8 @@ public class SlotSharingSlotAllocatorTest extends TestLogger {
final JobSchedulingPlan jobSchedulingPlan =
slotAllocator
- .determineParallelismAndCalculateAssignment(jobInformation, getSlots(50))
+ .determineParallelismAndCalculateAssignment(
+ jobInformation, getSlots(50), JobAllocationsInformation.empty())
.get();
final ReservedSlots reservedSlots =
@@ -234,7 +241,8 @@ public class SlotSharingSlotAllocatorTest extends TestLogger {
JobSchedulingPlan jobSchedulingPlan =
slotSharingSlotAllocator
- .determineParallelismAndCalculateAssignment(jobInformation, getSlots(50))
+ .determineParallelismAndCalculateAssignment(
+ jobInformation, getSlots(50), JobAllocationsInformation.empty())
.get();
final Optional<? extends ReservedSlots> reservedSlots =
@@ -243,6 +251,73 @@ public class SlotSharingSlotAllocatorTest extends TestLogger {
assertFalse(reservedSlots.isPresent());
}
+ /**
+ * Basic test to verify that allocation takes previous allocations into account to facilitate
+ * Local Recovery.
+ */
+ @Test
+ public void testStickyAllocation() {
+ Map<JobVertexID, List<VertexAllocationInformation>> locality = new HashMap<>();
+
+ // previous allocation allocation1: v1, v2
+ AllocationID allocation1 = new AllocationID();
+ locality.put(
+ vertex1.getJobVertexID(),
+ Collections.singletonList(
+ new VertexAllocationInformation(
+ allocation1, vertex1.getJobVertexID(), KeyGroupRange.of(1, 100))));
+ locality.put(
+ vertex2.getJobVertexID(),
+ Collections.singletonList(
+ new VertexAllocationInformation(
+ allocation1, vertex2.getJobVertexID(), KeyGroupRange.of(1, 100))));
+
+ // previous allocation allocation2: v3
+ AllocationID allocation2 = new AllocationID();
+ locality.put(
+ vertex3.getJobVertexID(),
+ Collections.singletonList(
+ new VertexAllocationInformation(
+ allocation2, vertex3.getJobVertexID(), KeyGroupRange.of(1, 100))));
+
+ List<SlotInfo> freeSlots = new ArrayList<>();
+ IntStream.range(0, 10).forEach(i -> freeSlots.add(new TestSlotInfo(new AllocationID())));
+ freeSlots.add(new TestSlotInfo(allocation1));
+ freeSlots.add(new TestSlotInfo(allocation2));
+
+ Map<JobVertexID, Long> stateSizes = new HashMap<>();
+ stateSizes.put(vertex1.getJobVertexID(), 10L);
+ stateSizes.put(vertex2.getJobVertexID(), 10L);
+ stateSizes.put(vertex3.getJobVertexID(), 10L);
+
+ JobSchedulingPlan schedulingPlan =
+ SlotSharingSlotAllocator.createSlotSharingSlotAllocator(
+ (allocationId, resourceProfile) ->
+ TestingPhysicalSlot.builder().build(),
+ (allocationID, cause, ts) -> {},
+ id -> false)
+ .determineParallelismAndCalculateAssignment(
+ new TestJobInformation(Arrays.asList(vertex1, vertex2, vertex3)),
+ freeSlots,
+ new JobAllocationsInformation(locality))
+ .get();
+
+ Map<AllocationID, Set<VertexID>> allocated = new HashMap<>();
+ for (JobSchedulingPlan.SlotAssignment assignment : schedulingPlan.getSlotAssignments()) {
+ ExecutionSlotSharingGroup target =
+ assignment.getTargetAs(ExecutionSlotSharingGroup.class);
+ Set<VertexID> set =
+ allocated.computeIfAbsent(
+ assignment.getSlotInfo().getAllocationId(), ign -> new HashSet<>());
+ for (ExecutionVertexID id : target.getContainedExecutionVertices()) {
+ set.add(id.getJobVertexId());
+ }
+ }
+ assertThat(allocated.get(allocation1)).contains(vertex1.getJobVertexID());
+ assertThat(allocated.get(allocation1)).contains(vertex2.getJobVertexID());
+ assertThat(allocated.get(allocation2)).contains(vertex3.getJobVertexID());
+ }
+
private static Collection<SlotInfo> getSlots(int count) {
final Collection<SlotInfo> slotInfo = new ArrayList<>();
for (int i = 0; i < count; i++) {
@@ -250,68 +325,4 @@ public class SlotSharingSlotAllocatorTest extends TestLogger {
}
return slotInfo;
}
-
- private static class TestJobInformation implements JobInformation {
-
- private final Map<JobVertexID, VertexInformation> vertexIdToInformation;
- private final Collection<SlotSharingGroup> slotSharingGroups;
-
- private TestJobInformation(Collection<VertexInformation> vertexIdToInformation) {
- this.vertexIdToInformation =
- vertexIdToInformation.stream()
- .collect(
- Collectors.toMap(
- VertexInformation::getJobVertexID,
- Function.identity()));
- this.slotSharingGroups =
- vertexIdToInformation.stream()
- .map(VertexInformation::getSlotSharingGroup)
- .collect(Collectors.toSet());
- }
-
- @Override
- public Collection<SlotSharingGroup> getSlotSharingGroups() {
- return slotSharingGroups;
- }
-
- @Override
- public VertexInformation getVertexInformation(JobVertexID jobVertexId) {
- return vertexIdToInformation.get(jobVertexId);
- }
-
- @Override
- public Iterable<VertexInformation> getVertices() {
- return vertexIdToInformation.values();
- }
- }
-
- private static class TestVertexInformation implements JobInformation.VertexInformation {
-
- private final JobVertexID jobVertexId;
- private final int parallelism;
- private final SlotSharingGroup slotSharingGroup;
-
- private TestVertexInformation(
- JobVertexID jobVertexId, int parallelism, SlotSharingGroup slotSharingGroup) {
- this.jobVertexId = jobVertexId;
- this.parallelism = parallelism;
- this.slotSharingGroup = slotSharingGroup;
- slotSharingGroup.addVertexToGroup(jobVertexId);
- }
-
- @Override
- public JobVertexID getJobVertexID() {
- return jobVertexId;
- }
-
- @Override
- public int getParallelism() {
- return parallelism;
- }
-
- @Override
- public SlotSharingGroup getSlotSharingGroup() {
- return slotSharingGroup;
- }
- }
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/StateLocalitySlotAssignerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/StateLocalitySlotAssignerTest.java
new file mode 100644
index 00000000000..aa9ea5a6bf9
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/StateLocalitySlotAssignerTest.java
@@ -0,0 +1,166 @@
+/*
+ * 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.adaptive.allocator;
+
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
+import org.apache.flink.runtime.scheduler.adaptive.JobSchedulingPlan.SlotAssignment;
+import org.apache.flink.runtime.scheduler.adaptive.allocator.JobAllocationsInformation.VertexAllocationInformation;
+import org.apache.flink.runtime.scheduler.adaptive.allocator.JobInformation.VertexInformation;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyGroupRangeAssignment;
+
+import org.hamcrest.MatcherAssert;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static java.util.Collections.singletonList;
+import static java.util.Collections.singletonMap;
+import static org.hamcrest.Matchers.hasItems;
+import static org.hamcrest.Matchers.hasSize;
+
+/** {@link StateLocalitySlotAssigner} test. */
+class StateLocalitySlotAssignerTest {
+ @Test
+ public void testSlotsAreNotWasted() {
+ VertexInformation vertex = createVertex(2);
+ AllocationID alloc1 = new AllocationID();
+ AllocationID alloc2 = new AllocationID();
+
+ List<VertexAllocationInformation> allocations =
+ Arrays.asList(
+ new VertexAllocationInformation(
+ alloc1, vertex.getJobVertexID(), KeyGroupRange.of(0, 9)),
+ new VertexAllocationInformation(
+ alloc2, vertex.getJobVertexID(), KeyGroupRange.of(10, 19)));
+
+ assign(vertex, Arrays.asList(alloc1, alloc2), allocations);
+ }
+
+ @Test
+ public void testUpScaling() {
+ final int oldParallelism = 3;
+ final int newParallelism = 7;
+ final int numFreeSlots = 100;
+ final VertexInformation vertex = createVertex(newParallelism);
+ final List<AllocationID> allocationIDs = createAllocationIDS(numFreeSlots);
+
+ List<VertexAllocationInformation> prevAllocations = new ArrayList<>();
+ Iterator<AllocationID> iterator = allocationIDs.iterator();
+ for (int i = 0; i < oldParallelism; i++) {
+ prevAllocations.add(
+ new VertexAllocationInformation(
+ iterator.next(),
+ vertex.getJobVertexID(),
+ KeyGroupRangeAssignment.computeKeyGroupRangeForOperatorIndex(
+ vertex.getMaxParallelism(), oldParallelism, i)));
+ }
+
+ Collection<SlotAssignment> assignments = assign(vertex, allocationIDs, prevAllocations);
+
+ verifyAssignments(
+ assignments,
+ newParallelism,
+ prevAllocations.stream()
+ .map(VertexAllocationInformation::getAllocationID)
+ .toArray(AllocationID[]::new));
+ }
+
+ @Test
+ public void testDownScaling() {
+ final int oldParallelism = 5;
+ final int newParallelism = 1;
+ final int numFreeSlots = 100;
+ final VertexInformation vertex = createVertex(newParallelism);
+ final List<AllocationID> allocationIDs = createAllocationIDS(numFreeSlots);
+
+ // pretend that the 1st (0) subtask had half of key groups ...
+ final Iterator<AllocationID> iterator = allocationIDs.iterator();
+ final AllocationID biggestAllocation = iterator.next();
+ final List<VertexAllocationInformation> prevAllocations = new ArrayList<>();
+ final int halfOfKeyGroupRange = vertex.getMaxParallelism() / 2;
+ prevAllocations.add(
+ new VertexAllocationInformation(
+ biggestAllocation,
+ vertex.getJobVertexID(),
+ KeyGroupRange.of(0, halfOfKeyGroupRange - 1)));
+
+ // and the remaining subtasks had only one key group each
+ for (int subtaskIdx = 1; subtaskIdx < oldParallelism; subtaskIdx++) {
+ int keyGroup = halfOfKeyGroupRange + subtaskIdx;
+ prevAllocations.add(
+ new VertexAllocationInformation(
+ iterator.next(),
+ vertex.getJobVertexID(),
+ KeyGroupRange.of(keyGroup, keyGroup)));
+ }
+
+ Collection<SlotAssignment> assignments = assign(vertex, allocationIDs, prevAllocations);
+
+ verifyAssignments(assignments, newParallelism, biggestAllocation);
+ }
+
+ private static void verifyAssignments(
+ Collection<SlotAssignment> assignments,
+ int expectedSize,
+ AllocationID... mustHaveAllocationID) {
+ MatcherAssert.assertThat(assignments, hasSize(expectedSize));
+ MatcherAssert.assertThat(
+ assignments.stream()
+ .map(e -> e.getSlotInfo().getAllocationId())
+ .collect(Collectors.toSet()),
+ hasItems(mustHaveAllocationID));
+ }
+
+ private static Collection<SlotAssignment> assign(
+ VertexInformation vertexInformation,
+ List<AllocationID> allocationIDs,
+ List<VertexAllocationInformation> allocations) {
+ return new StateLocalitySlotAssigner()
+ .assignSlots(
+ new TestJobInformation(singletonList(vertexInformation)),
+ allocationIDs.stream().map(TestSlotInfo::new).collect(Collectors.toList()),
+ new VertexParallelism(
+ singletonMap(
+ vertexInformation.getJobVertexID(),
+ vertexInformation.getParallelism())),
+ new JobAllocationsInformation(
+ singletonMap(vertexInformation.getJobVertexID(), allocations)));
+ }
+
+ private static VertexInformation createVertex(int parallelism) {
+ JobVertexID id = new JobVertexID();
+ SlotSharingGroup slotSharingGroup = new SlotSharingGroup();
+ slotSharingGroup.addVertexToGroup(id);
+ return new TestVertexInformation(id, parallelism, slotSharingGroup);
+ }
+
+ private static List<AllocationID> createAllocationIDS(int numFreeSlots) {
+ return IntStream.range(0, numFreeSlots)
+ .mapToObj(i -> new AllocationID())
+ .collect(Collectors.toList());
+ }
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/TestJobInformation.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/TestJobInformation.java
new file mode 100644
index 00000000000..b9a3671d3c4
--- /dev/null
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/TestJobInformation.java
@@ -0,0 +1,59 @@
+/*
+ * 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.adaptive.allocator;
+
+import org.apache.flink.runtime.jobgraph.JobVertexID;
+import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+class TestJobInformation implements JobInformation {
+
+ private final Map<JobVertexID, VertexInformation> vertexIdToInformation;
+ private final Collection<SlotSharingGroup> slotSharingGroups;
+
+ TestJobInformation(Collection<? extends VertexInformation> vertexIdToInformation) {
+ this.vertexIdToInformation =
+ vertexIdToInformation.stream()
+ .collect(
+ Collectors.toMap(
+ VertexInformation::getJobVertexID, Function.identity()));
+ this.slotSharingGroups =
+ vertexIdToInformation.stream()
+ .map(VertexInformation::getSlotSharingGroup)
+ .collect(Collectors.toSet());
+ }
+
+ @Override
+ public Collection<SlotSharingGroup> getSlotSharingGroups() {
+ return slotSharingGroups;
+ }
+
+ @Override
+ public VertexInformation getVertexInformation(JobVertexID jobVertexId) {
+ return vertexIdToInformation.get(jobVertexId);
+ }
+
+ @Override
+ public Iterable<VertexInformation> getVertices() {
+ return vertexIdToInformation.values();
+ }
+}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/TestSlotInfo.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/TestSlotInfo.java
index b023e5edf49..e467554ddfa 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/TestSlotInfo.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/TestSlotInfo.java
@@ -26,7 +26,15 @@ import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
/** Test {@link SlotInfo} implementation. */
class TestSlotInfo implements SlotInfo {
- private final AllocationID allocationId = new AllocationID();
+ private final AllocationID allocationId;
+
+ public TestSlotInfo() {
+ this(new AllocationID());
+ }
+
+ public TestSlotInfo(AllocationID allocationId) {
+ this.allocationId = allocationId;
+ }
@Override
public AllocationID getAllocationId() {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/JobInformation.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/TestVertexInformation.java
similarity index 53%
copy from flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/JobInformation.java
copy to flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/TestVertexInformation.java
index 99b8d48c051..4666ba20647 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/JobInformation.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/TestVertexInformation.java
@@ -20,30 +20,37 @@ package org.apache.flink.runtime.scheduler.adaptive.allocator;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
-import java.util.Collection;
-
-/** Information about the job. */
-public interface JobInformation {
- /**
- * Returns all slot-sharing groups of the job.
- *
- * <p>Attention: The returned slot sharing groups should never be modified (they are indeed
- * mutable)!
- *
- * @return all slot-sharing groups of the job
- */
- Collection<SlotSharingGroup> getSlotSharingGroups();
-
- VertexInformation getVertexInformation(JobVertexID jobVertexId);
+class TestVertexInformation implements JobInformation.VertexInformation {
+
+ private final JobVertexID jobVertexId;
+ private final int parallelism;
+ private final SlotSharingGroup slotSharingGroup;
+
+ TestVertexInformation(
+ JobVertexID jobVertexId, int parallelism, SlotSharingGroup slotSharingGroup) {
+ this.jobVertexId = jobVertexId;
+ this.parallelism = parallelism;
+ this.slotSharingGroup = slotSharingGroup;
+ slotSharingGroup.addVertexToGroup(jobVertexId);
+ }
- Iterable<VertexInformation> getVertices();
+ @Override
+ public JobVertexID getJobVertexID() {
+ return jobVertexId;
+ }
- /** Information about a single vertex. */
- interface VertexInformation {
- JobVertexID getJobVertexID();
+ @Override
+ public int getParallelism() {
+ return parallelism;
+ }
- int getParallelism();
+ @Override
+ public int getMaxParallelism() {
+ return 128;
+ }
- SlotSharingGroup getSlotSharingGroup();
+ @Override
+ public SlotSharingGroup getSlotSharingGroup() {
+ return slotSharingGroup;
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/TestingSlotAllocator.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/TestingSlotAllocator.java
index 44b73122ae9..0fd638c43bf 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/TestingSlotAllocator.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/TestingSlotAllocator.java
@@ -24,7 +24,6 @@ import org.apache.flink.runtime.util.ResourceCounter;
import java.util.Collection;
import java.util.Optional;
-import java.util.function.BiFunction;
import java.util.function.Function;
/** Testing implementation of {@link SlotAllocator}. */
@@ -33,20 +32,13 @@ public class TestingSlotAllocator implements SlotAllocator {
private final Function<Iterable<JobInformation.VertexInformation>, ResourceCounter>
calculateRequiredSlotsFunction;
- private final BiFunction<
- JobInformation, Collection<? extends SlotInfo>, Optional<VertexParallelism>>
- determineParallelismFunction;
-
private final Function<VertexParallelism, Optional<ReservedSlots>> tryReserveResourcesFunction;
private TestingSlotAllocator(
Function<Iterable<JobInformation.VertexInformation>, ResourceCounter>
calculateRequiredSlotsFunction,
- BiFunction<JobInformation, Collection<? extends SlotInfo>, Optional<VertexParallelism>>
- determineParallelismFunction,
Function<VertexParallelism, Optional<ReservedSlots>> tryReserveResourcesFunction) {
this.calculateRequiredSlotsFunction = calculateRequiredSlotsFunction;
- this.determineParallelismFunction = determineParallelismFunction;
this.tryReserveResourcesFunction = tryReserveResourcesFunction;
}
@@ -59,12 +51,14 @@ public class TestingSlotAllocator implements SlotAllocator {
@Override
public Optional<VertexParallelism> determineParallelism(
JobInformation jobInformation, Collection<? extends SlotInfo> slots) {
- return determineParallelismFunction.apply(jobInformation, slots);
+ return Optional.empty();
}
@Override
public Optional<JobSchedulingPlan> determineParallelismAndCalculateAssignment(
- JobInformation jobInformation, Collection<? extends SlotInfo> slots) {
+ JobInformation jobInformation,
+ Collection<? extends SlotInfo> slots,
+ JobAllocationsInformation jobAllocationsInformation) {
return Optional.empty();
}
@@ -81,9 +75,6 @@ public class TestingSlotAllocator implements SlotAllocator {
public static final class Builder {
private Function<Iterable<JobInformation.VertexInformation>, ResourceCounter>
calculateRequiredSlotsFunction = ignored -> ResourceCounter.empty();
- private BiFunction<
- JobInformation, Collection<? extends SlotInfo>, Optional<VertexParallelism>>
- determineParallelismFunction = (ignoredA, ignoredB) -> Optional.empty();
private Function<VertexParallelism, Optional<ReservedSlots>> tryReserveResourcesFunction =
ignored -> Optional.empty();
@@ -94,16 +85,6 @@ public class TestingSlotAllocator implements SlotAllocator {
return this;
}
- public Builder setDetermineParallelismFunction(
- BiFunction<
- JobInformation,
- Collection<? extends SlotInfo>,
- Optional<VertexParallelism>>
- determineParallelismFunction) {
- this.determineParallelismFunction = determineParallelismFunction;
- return this;
- }
-
public Builder setTryReserveResourcesFunction(
Function<VertexParallelism, Optional<ReservedSlots>> tryReserveResourcesFunction) {
this.tryReserveResourcesFunction = tryReserveResourcesFunction;
@@ -112,9 +93,7 @@ public class TestingSlotAllocator implements SlotAllocator {
public TestingSlotAllocator build() {
return new TestingSlotAllocator(
- calculateRequiredSlotsFunction,
- determineParallelismFunction,
- tryReserveResourcesFunction);
+ calculateRequiredSlotsFunction, tryReserveResourcesFunction);
}
}
}
[flink] 02/04: [FLINK-21450][runtime] Add previous ExecutionGraph to WaitingForResources AdaptiveScheduler state
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
roman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 676e8e5528f99eb8ba5747f7489b0f02ee025dd6
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Mon Feb 20 23:08:47 2023 +0000
[FLINK-21450][runtime] Add previous ExecutionGraph to WaitingForResources AdaptiveScheduler state
Previous ExecutionGraph will be used in a subsequent commit to allocate
workloads more optimally by taking previous allocations into account.
---
.../scheduler/adaptive/AdaptiveScheduler.java | 12 ++++++----
.../flink/runtime/scheduler/adaptive/Created.java | 2 +-
.../scheduler/adaptive/CreatingExecutionGraph.java | 17 +++++++++----
.../runtime/scheduler/adaptive/Restarting.java | 5 +++-
.../scheduler/adaptive/StateTransitions.java | 6 +++--
.../scheduler/adaptive/WaitingForResources.java | 28 +++++++++++++++++-----
.../runtime/scheduler/adaptive/CreatedTest.java | 3 ++-
.../adaptive/CreatingExecutionGraphTest.java | 23 +++++++++++-------
.../runtime/scheduler/adaptive/RestartingTest.java | 2 +-
.../adaptive/WaitingForResourcesTest.java | 9 ++++---
10 files changed, 76 insertions(+), 31 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
index 4f7b559e86a..0bdbb44da7c 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
@@ -784,7 +784,7 @@ public class AdaptiveScheduler
}
@Override
- public void goToWaitingForResources() {
+ public void goToWaitingForResources(@Nullable ExecutionGraph previousExecutionGraph) {
final ResourceCounter desiredResources = calculateDesiredResources();
declarativeSlotPool.setResourceRequirements(desiredResources);
@@ -794,7 +794,8 @@ public class AdaptiveScheduler
LOG,
desiredResources,
this.initialResourceAllocationTimeout,
- this.resourceStabilizationTimeout));
+ this.resourceStabilizationTimeout,
+ previousExecutionGraph));
}
private ResourceCounter calculateDesiredResources() {
@@ -916,14 +917,17 @@ public class AdaptiveScheduler
}
@Override
- public void goToCreatingExecutionGraph() {
+ public void goToCreatingExecutionGraph(@Nullable ExecutionGraph previousExecutionGraph) {
final CompletableFuture<CreatingExecutionGraph.ExecutionGraphWithVertexParallelism>
executionGraphWithAvailableResourcesFuture =
createExecutionGraphWithAvailableResourcesAsync();
transitionToState(
new CreatingExecutionGraph.Factory(
- this, executionGraphWithAvailableResourcesFuture, LOG));
+ this,
+ executionGraphWithAvailableResourcesFuture,
+ LOG,
+ previousExecutionGraph));
}
private CompletableFuture<CreatingExecutionGraph.ExecutionGraphWithVertexParallelism>
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Created.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Created.java
index 1dd9f1e8f16..798eeda352e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Created.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Created.java
@@ -69,7 +69,7 @@ class Created implements State {
/** Starts the scheduling by going into the {@link WaitingForResources} state. */
void startScheduling() {
- context.goToWaitingForResources();
+ context.goToWaitingForResources(null);
}
/** Context of the {@link Created} state. */
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraph.java
index 0ec7437bb85..c87e58b2971 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraph.java
@@ -61,12 +61,15 @@ public class CreatingExecutionGraph implements State {
private final Logger logger;
private final OperatorCoordinatorHandlerFactory operatorCoordinatorHandlerFactory;
+ private final @Nullable ExecutionGraph previousExecutionGraph;
+
public CreatingExecutionGraph(
Context context,
CompletableFuture<ExecutionGraphWithVertexParallelism>
executionGraphWithParallelismFuture,
Logger logger,
- OperatorCoordinatorHandlerFactory operatorCoordinatorFactory) {
+ OperatorCoordinatorHandlerFactory operatorCoordinatorFactory,
+ ExecutionGraph previousExecutionGraph1) {
this.context = context;
this.logger = logger;
this.operatorCoordinatorHandlerFactory = operatorCoordinatorFactory;
@@ -82,6 +85,7 @@ public class CreatingExecutionGraph implements State {
Duration.ZERO);
return null;
}));
+ previousExecutionGraph = previousExecutionGraph1;
}
private void handleExecutionGraphCreation(
@@ -141,7 +145,7 @@ public class CreatingExecutionGraph implements State {
} else {
logger.debug(
"Failed to reserve and assign the required slots. Waiting for new resources.");
- context.goToWaitingForResources();
+ context.goToWaitingForResources(previousExecutionGraph);
}
}
}
@@ -300,16 +304,20 @@ public class CreatingExecutionGraph implements State {
private final CompletableFuture<ExecutionGraphWithVertexParallelism>
executionGraphWithParallelismFuture;
+ private final @Nullable ExecutionGraph previousExecutionGraph;
+
private final Logger log;
Factory(
Context context,
CompletableFuture<ExecutionGraphWithVertexParallelism>
executionGraphWithParallelismFuture,
- Logger log) {
+ Logger log,
+ @Nullable ExecutionGraph previousExecutionGraph) {
this.context = context;
this.executionGraphWithParallelismFuture = executionGraphWithParallelismFuture;
this.log = log;
+ this.previousExecutionGraph = previousExecutionGraph;
}
@Override
@@ -323,7 +331,8 @@ public class CreatingExecutionGraph implements State {
context,
executionGraphWithParallelismFuture,
log,
- DefaultOperatorCoordinatorHandler::new);
+ DefaultOperatorCoordinatorHandler::new,
+ previousExecutionGraph);
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Restarting.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Restarting.java
index 125804959dc..f0a38035f44 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Restarting.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/Restarting.java
@@ -97,7 +97,10 @@ class Restarting extends StateWithExecutionGraph {
void onGloballyTerminalState(JobStatus globallyTerminalState) {
Preconditions.checkArgument(globallyTerminalState == JobStatus.CANCELED);
goToWaitingForResourcesFuture =
- context.runIfState(this, context::goToWaitingForResources, backoffTime);
+ context.runIfState(
+ this,
+ () -> context.goToWaitingForResources(getExecutionGraph()),
+ backoffTime);
}
/** Context of the {@link Restarting} state. */
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitions.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitions.java
index bd3f8a509c4..f6881be4dde 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitions.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/StateTransitions.java
@@ -25,6 +25,8 @@ import org.apache.flink.runtime.scheduler.ExecutionGraphHandler;
import org.apache.flink.runtime.scheduler.OperatorCoordinatorHandler;
import org.apache.flink.runtime.scheduler.exceptionhistory.ExceptionHistoryEntry;
+import javax.annotation.Nullable;
+
import java.time.Duration;
import java.util.List;
import java.util.concurrent.CompletableFuture;
@@ -58,7 +60,7 @@ public interface StateTransitions {
interface ToCreatingExecutionGraph extends StateTransitions {
/** Transitions into the {@link CreatingExecutionGraph} state. */
- void goToCreatingExecutionGraph();
+ void goToCreatingExecutionGraph(@Nullable ExecutionGraph previousExecutionGraph);
}
/** Interface covering transition to the {@link Executing} state. */
@@ -164,6 +166,6 @@ public interface StateTransitions {
interface ToWaitingForResources extends StateTransitions {
/** Transitions into the {@link WaitingForResources} state. */
- void goToWaitingForResources();
+ void goToWaitingForResources(@Nullable ExecutionGraph previousExecutionGraph);
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResources.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResources.java
index 7081bb8980f..f102bcfc4ba 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResources.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResources.java
@@ -22,6 +22,7 @@ import org.apache.flink.annotation.VisibleForTesting;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.api.common.time.Deadline;
import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
+import org.apache.flink.runtime.executiongraph.ExecutionGraph;
import org.apache.flink.runtime.util.ResourceCounter;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.clock.Clock;
@@ -53,6 +54,9 @@ class WaitingForResources implements State, ResourceConsumer {
@Nullable private ScheduledFuture<?> resourceTimeoutFuture;
+ @Nullable private final ExecutionGraph previousExecutionGraph;
+
+ @VisibleForTesting
WaitingForResources(
Context context,
Logger log,
@@ -65,17 +69,18 @@ class WaitingForResources implements State, ResourceConsumer {
desiredResources,
initialResourceAllocationTimeout,
resourceStabilizationTimeout,
- SystemClock.getInstance());
+ SystemClock.getInstance(),
+ null);
}
- @VisibleForTesting
WaitingForResources(
Context context,
Logger log,
ResourceCounter desiredResources,
Duration initialResourceAllocationTimeout,
Duration resourceStabilizationTimeout,
- Clock clock) {
+ Clock clock,
+ @Nullable ExecutionGraph previousExecutionGraph) {
this.context = Preconditions.checkNotNull(context);
this.log = Preconditions.checkNotNull(log);
this.desiredResources = Preconditions.checkNotNull(desiredResources);
@@ -97,6 +102,7 @@ class WaitingForResources implements State, ResourceConsumer {
context.runIfState(
this, this::resourceTimeout, initialResourceAllocationTimeout);
}
+ this.previousExecutionGraph = previousExecutionGraph;
context.runIfState(this, this::notifyNewResourcesAvailable, Duration.ZERO);
}
@@ -175,7 +181,7 @@ class WaitingForResources implements State, ResourceConsumer {
}
private void createExecutionGraphWithAvailableResources() {
- context.goToCreatingExecutionGraph();
+ context.goToCreatingExecutionGraph(previousExecutionGraph);
}
/** Context of the {@link WaitingForResources} state. */
@@ -227,18 +233,21 @@ class WaitingForResources implements State, ResourceConsumer {
private final ResourceCounter desiredResources;
private final Duration initialResourceAllocationTimeout;
private final Duration resourceStabilizationTimeout;
+ @Nullable private final ExecutionGraph previousExecutionGraph;
public Factory(
Context context,
Logger log,
ResourceCounter desiredResources,
Duration initialResourceAllocationTimeout,
- Duration resourceStabilizationTimeout) {
+ Duration resourceStabilizationTimeout,
+ ExecutionGraph previousExecutionGraph) {
this.context = context;
this.log = log;
this.desiredResources = desiredResources;
this.initialResourceAllocationTimeout = initialResourceAllocationTimeout;
this.resourceStabilizationTimeout = resourceStabilizationTimeout;
+ this.previousExecutionGraph = previousExecutionGraph;
}
public Class<WaitingForResources> getStateClass() {
@@ -251,7 +260,14 @@ class WaitingForResources implements State, ResourceConsumer {
log,
desiredResources,
initialResourceAllocationTimeout,
- resourceStabilizationTimeout);
+ resourceStabilizationTimeout,
+ SystemClock.getInstance(),
+ previousExecutionGraph);
}
}
+
+ @Nullable
+ public ExecutionGraph getPreviousExecutionGraph() {
+ return previousExecutionGraph;
+ }
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatedTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatedTest.java
index 6887308f01b..e63513f16df 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatedTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatedTest.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.scheduler.adaptive;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.JobStatus;
import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
+import org.apache.flink.runtime.executiongraph.ExecutionGraph;
import org.apache.flink.util.TestLogger;
import org.junit.Test;
@@ -122,7 +123,7 @@ public class CreatedTest extends TestLogger {
}
@Override
- public void goToWaitingForResources() {
+ public void goToWaitingForResources(@Nullable ExecutionGraph previousExecutionGraph) {
waitingForResourcesStateValidator.validateInput(null);
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraphTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraphTest.java
index aa92d1dc089..85a4fe4f3ab 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraphTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraphTest.java
@@ -67,7 +67,8 @@ public class CreatingExecutionGraphTest extends TestLogger {
context,
new CompletableFuture<>(),
log,
- CreatingExecutionGraphTest::createTestingOperatorCoordinatorHandler);
+ CreatingExecutionGraphTest::createTestingOperatorCoordinatorHandler,
+ null);
context.setExpectFinished(
archivedExecutionGraph ->
@@ -86,7 +87,8 @@ public class CreatingExecutionGraphTest extends TestLogger {
context,
new CompletableFuture<>(),
log,
- CreatingExecutionGraphTest::createTestingOperatorCoordinatorHandler);
+ CreatingExecutionGraphTest::createTestingOperatorCoordinatorHandler,
+ null);
context.setExpectFinished(
archivedExecutionGraph ->
@@ -105,7 +107,8 @@ public class CreatingExecutionGraphTest extends TestLogger {
context,
new CompletableFuture<>(),
log,
- CreatingExecutionGraphTest::createTestingOperatorCoordinatorHandler);
+ CreatingExecutionGraphTest::createTestingOperatorCoordinatorHandler,
+ null);
context.setExpectFinished(
archivedExecutionGraph ->
@@ -125,7 +128,8 @@ public class CreatingExecutionGraphTest extends TestLogger {
context,
executionGraphWithVertexParallelismFuture,
log,
- CreatingExecutionGraphTest::createTestingOperatorCoordinatorHandler);
+ CreatingExecutionGraphTest::createTestingOperatorCoordinatorHandler,
+ null);
context.setExpectFinished(
archivedExecutionGraph ->
@@ -146,7 +150,8 @@ public class CreatingExecutionGraphTest extends TestLogger {
context,
executionGraphWithVertexParallelismFuture,
log,
- CreatingExecutionGraphTest::createTestingOperatorCoordinatorHandler);
+ CreatingExecutionGraphTest::createTestingOperatorCoordinatorHandler,
+ null);
context.setTryToAssignSlotsFunction(
ignored -> CreatingExecutionGraph.AssignmentResult.notPossible());
@@ -167,7 +172,8 @@ public class CreatingExecutionGraphTest extends TestLogger {
context,
executionGraphWithVertexParallelismFuture,
log,
- CreatingExecutionGraphTest::createTestingOperatorCoordinatorHandler);
+ CreatingExecutionGraphTest::createTestingOperatorCoordinatorHandler,
+ null);
final StateTrackingMockExecutionGraph executionGraph =
new StateTrackingMockExecutionGraph();
@@ -197,7 +203,8 @@ public class CreatingExecutionGraphTest extends TestLogger {
(executionGraph, errorHandler) -> {
operatorCoordinatorGlobalFailureHandlerRef.set(errorHandler);
return new TestingOperatorCoordinatorHandler();
- });
+ },
+ null);
final StateTrackingMockExecutionGraph executionGraph =
new StateTrackingMockExecutionGraph();
@@ -318,7 +325,7 @@ public class CreatingExecutionGraphTest extends TestLogger {
}
@Override
- public void goToWaitingForResources() {
+ public void goToWaitingForResources(@Nullable ExecutionGraph previousExecutionGraph) {
waitingForResourcesStateValidator.validateInput(null);
hadStateTransitionHappened = true;
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/RestartingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/RestartingTest.java
index 94e7b75992d..264cb221f70 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/RestartingTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/RestartingTest.java
@@ -174,7 +174,7 @@ public class RestartingTest extends TestLogger {
public void archiveFailure(RootExceptionHistoryEntry failure) {}
@Override
- public void goToWaitingForResources() {
+ public void goToWaitingForResources(ExecutionGraph previousExecutionGraph) {
waitingForResourcesStateValidator.validateInput(null);
hadStateTransition = true;
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResourcesTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResourcesTest.java
index 504a29e65ec..2c58f053b13 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResourcesTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/WaitingForResourcesTest.java
@@ -23,6 +23,7 @@ import org.apache.flink.core.testutils.ScheduledTask;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
import org.apache.flink.runtime.executiongraph.ErrorInfo;
+import org.apache.flink.runtime.executiongraph.ExecutionGraph;
import org.apache.flink.runtime.rest.handler.legacy.utils.ArchivedExecutionGraphBuilder;
import org.apache.flink.runtime.util.ResourceCounter;
import org.apache.flink.util.TestLogger;
@@ -157,7 +158,8 @@ public class WaitingForResourcesTest extends TestLogger {
RESOURCE_COUNTER,
initialResourceTimeout,
stabilizationTimeout,
- ctx.getClock());
+ ctx.getClock(),
+ null);
// sufficient resources available
ctx.setHasDesiredResources(() -> false);
ctx.setHasSufficientResources(() -> true);
@@ -191,7 +193,8 @@ public class WaitingForResourcesTest extends TestLogger {
RESOURCE_COUNTER,
initialResourceTimeout,
stabilizationTimeout,
- ctx.getClock());
+ ctx.getClock(),
+ null);
ctx.setHasDesiredResources(() -> false);
@@ -515,7 +518,7 @@ public class WaitingForResourcesTest extends TestLogger {
}
@Override
- public void goToCreatingExecutionGraph() {
+ public void goToCreatingExecutionGraph(@Nullable ExecutionGraph previousExecutionGraph) {
creatingExecutionGraphStateValidator.validateInput(null);
hasStateTransition = true;
}
[flink] 03/04: [FLINK-21450][runtime] Restructure types passed between AdaptiveScheduler and SlotAssigner
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
roman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 416cb7aaa02c176e01485ff11ab4269f76b5e9e2
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Tue Feb 28 14:22:28 2023 +0000
[FLINK-21450][runtime] Restructure types passed between AdaptiveScheduler and SlotAssigner
Slot assignments are computed and consumed by SlotAllocator.
This is expressed implicitly by extending VertexParallelism.
This change tries to make that clear, while still allowing to assign
slots to something other than Slot Sharing Groups.
It does so by:
1. Introduce JobSchedulingPlan, computed and consumed by SlotAllocator. It couples VertexParallelism with slot assignments
2. Introduce determineParallelismAndCalculateAssignment method in addition to determineParallelism, specifically for assignments
3. Push the polymorphism of state assignments from VertexParallelism into the JobSchedulingPlan (slot assignment target)
---
.../jobgraph/jsonplan/JsonPlanGenerator.java | 21 +---
.../scheduler/adaptive/AdaptiveScheduler.java | 28 +++---
.../scheduler/adaptive/CreatingExecutionGraph.java | 14 ++-
.../scheduler/adaptive/JobSchedulingPlan.java | 105 +++++++++++++++++++
.../adaptive/allocator/SlotAllocator.java | 16 ++-
.../allocator/SlotSharingSlotAllocator.java | 111 +++++++++------------
.../adaptive/allocator/VertexParallelism.java | 43 ++++++--
.../VertexParallelismWithSlotSharing.java | 52 ----------
.../scheduler/adaptive/AdaptiveSchedulerTest.java | 3 +-
.../adaptive/CreatingExecutionGraphTest.java | 29 ++----
.../allocator/SlotSharingSlotAllocatorTest.java | 62 ++++++------
.../adaptive/allocator/TestingSlotAllocator.java | 28 +++---
12 files changed, 276 insertions(+), 236 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/jsonplan/JsonPlanGenerator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/jsonplan/JsonPlanGenerator.java
index 8817c72282b..6cdcc77cc53 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/jsonplan/JsonPlanGenerator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobgraph/jsonplan/JsonPlanGenerator.java
@@ -33,27 +33,13 @@ import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator
import org.apache.commons.text.StringEscapeUtils;
import java.io.StringWriter;
-import java.util.Collections;
import java.util.List;
-import java.util.Map;
@Internal
public class JsonPlanGenerator {
private static final String NOT_SET = "";
private static final String EMPTY = "{}";
- private static final VertexParallelism EMPTY_VERTEX_PARALLELISM =
- new VertexParallelism() {
- @Override
- public Map<JobVertexID, Integer> getMaxParallelismForVertices() {
- return Collections.emptyMap();
- }
-
- @Override
- public int getParallelism(JobVertexID jobVertexId) {
- return -1;
- }
- };
public static String generatePlan(JobGraph jg) {
return generatePlan(
@@ -61,7 +47,7 @@ public class JsonPlanGenerator {
jg.getName(),
jg.getJobType(),
jg.getVertices(),
- EMPTY_VERTEX_PARALLELISM);
+ VertexParallelism.empty());
}
public static String generatePlan(
@@ -116,11 +102,12 @@ public class JsonPlanGenerator {
// write the core properties
JobVertexID vertexID = vertex.getID();
- int storeParallelism = vertexParallelism.getParallelism(vertexID);
gen.writeStringField("id", vertexID.toString());
gen.writeNumberField(
"parallelism",
- storeParallelism != -1 ? storeParallelism : vertex.getParallelism());
+ vertexParallelism
+ .getParallelismOptional(vertexID)
+ .orElse(vertex.getParallelism()));
gen.writeStringField("operator", operator);
gen.writeStringField("operator_strategy", operatorDescr);
gen.writeStringField("description", description);
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
index 0bdbb44da7c..a876bb65eb2 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveScheduler.java
@@ -760,11 +760,12 @@ public class AdaptiveScheduler
.isPresent();
}
- private VertexParallelism determineParallelism(SlotAllocator slotAllocator)
+ private JobSchedulingPlan determineParallelism(SlotAllocator slotAllocator)
throws NoResourceAvailableException {
return slotAllocator
- .determineParallelism(jobInformation, declarativeSlotPool.getFreeSlotsInformation())
+ .determineParallelismAndCalculateAssignment(
+ jobInformation, declarativeSlotPool.getFreeSlotsInformation())
.orElseThrow(
() ->
new NoResourceAvailableException(
@@ -932,11 +933,11 @@ public class AdaptiveScheduler
private CompletableFuture<CreatingExecutionGraph.ExecutionGraphWithVertexParallelism>
createExecutionGraphWithAvailableResourcesAsync() {
- final VertexParallelism vertexParallelism;
+ final JobSchedulingPlan schedulingPlan;
final VertexParallelismStore adjustedParallelismStore;
try {
- vertexParallelism = determineParallelism(slotAllocator);
+ schedulingPlan = determineParallelism(slotAllocator);
JobGraph adjustedJobGraph = jobInformation.copyJobGraph();
for (JobVertex vertex : adjustedJobGraph.getVertices()) {
@@ -944,7 +945,7 @@ public class AdaptiveScheduler
// use the determined "available parallelism" to use
// the resources we have access to
- vertex.setParallelism(vertexParallelism.getParallelism(id));
+ vertex.setParallelism(schedulingPlan.getVertexParallelism().getParallelism(id));
}
// use the originally configured max parallelism
@@ -966,7 +967,7 @@ public class AdaptiveScheduler
.thenApply(
executionGraph ->
CreatingExecutionGraph.ExecutionGraphWithVertexParallelism.create(
- executionGraph, vertexParallelism));
+ executionGraph, schedulingPlan));
}
@Override
@@ -982,10 +983,10 @@ public class AdaptiveScheduler
executionGraph.setInternalTaskFailuresListener(
new UpdateSchedulerNgOnInternalFailuresListener(this));
- final VertexParallelism vertexParallelism =
- executionGraphWithVertexParallelism.getVertexParallelism();
+ final JobSchedulingPlan jobSchedulingPlan =
+ executionGraphWithVertexParallelism.getJobSchedulingPlan();
return slotAllocator
- .tryReserveResources(vertexParallelism)
+ .tryReserveResources(jobSchedulingPlan)
.map(
reservedSlots ->
CreatingExecutionGraph.AssignmentResult.success(
@@ -1050,14 +1051,14 @@ public class AdaptiveScheduler
int availableSlots = declarativeSlotPool.getFreeSlotsInformation().size();
if (availableSlots > 0) {
- final Optional<? extends VertexParallelism> potentialNewParallelism =
+ final Optional<VertexParallelism> potentialNewParallelism =
slotAllocator.determineParallelism(
jobInformation, declarativeSlotPool.getAllSlotsInformation());
if (potentialNewParallelism.isPresent()) {
int currentCumulativeParallelism = getCurrentCumulativeParallelism(executionGraph);
int newCumulativeParallelism =
- getCumulativeParallelism(potentialNewParallelism.get());
+ potentialNewParallelism.get().getCumulativeParallelism();
if (newCumulativeParallelism > currentCumulativeParallelism) {
LOG.debug(
"Offering scale up to scale up controller with currentCumulativeParallelism={}, newCumulativeParallelism={}",
@@ -1077,11 +1078,6 @@ public class AdaptiveScheduler
.reduce(0, Integer::sum);
}
- private static int getCumulativeParallelism(VertexParallelism potentialNewParallelism) {
- return potentialNewParallelism.getMaxParallelismForVertices().values().stream()
- .reduce(0, Integer::sum);
- }
-
@Override
public void onFinished(ArchivedExecutionGraph archivedExecutionGraph) {
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraph.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraph.java
index c87e58b2971..f19145f0844 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraph.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraph.java
@@ -339,16 +339,16 @@ public class CreatingExecutionGraph implements State {
static class ExecutionGraphWithVertexParallelism {
private final ExecutionGraph executionGraph;
- private final VertexParallelism vertexParallelism;
+ private final JobSchedulingPlan jobSchedulingPlan;
private ExecutionGraphWithVertexParallelism(
- ExecutionGraph executionGraph, VertexParallelism vertexParallelism) {
+ ExecutionGraph executionGraph, JobSchedulingPlan jobSchedulingPlan) {
this.executionGraph = executionGraph;
- this.vertexParallelism = vertexParallelism;
+ this.jobSchedulingPlan = jobSchedulingPlan;
}
public static ExecutionGraphWithVertexParallelism create(
- ExecutionGraph executionGraph, VertexParallelism vertexParallelism) {
+ ExecutionGraph executionGraph, JobSchedulingPlan vertexParallelism) {
return new ExecutionGraphWithVertexParallelism(executionGraph, vertexParallelism);
}
@@ -357,7 +357,11 @@ public class CreatingExecutionGraph implements State {
}
public VertexParallelism getVertexParallelism() {
- return vertexParallelism;
+ return jobSchedulingPlan.getVertexParallelism();
+ }
+
+ public JobSchedulingPlan getJobSchedulingPlan() {
+ return jobSchedulingPlan;
}
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/JobSchedulingPlan.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/JobSchedulingPlan.java
new file mode 100644
index 00000000000..cde53948d16
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/JobSchedulingPlan.java
@@ -0,0 +1,105 @@
+/*
+ * 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.adaptive;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.jobmaster.SlotInfo;
+import org.apache.flink.runtime.scheduler.adaptive.allocator.VertexParallelism;
+
+import java.util.Collection;
+import java.util.Collections;
+
+/**
+ * A plan that describes how to execute {@link org.apache.flink.runtime.jobgraph.JobGraph JobGraph}.
+ *
+ * <ol>
+ * <li>{@link #vertexParallelism} is necessary to create {@link
+ * org.apache.flink.runtime.executiongraph.ExecutionGraph ExecutionGraph}
+ * <li>{@link #slotAssignments} are used to schedule it onto the cluster
+ * </ol>
+ *
+ * {@link AdaptiveScheduler} passes this structure from {@link WaitingForResources} to {@link
+ * CreatingExecutionGraph} stages.
+ */
+@Internal
+public class JobSchedulingPlan {
+ private final VertexParallelism vertexParallelism;
+ private final Collection<SlotAssignment> slotAssignments;
+
+ public JobSchedulingPlan(
+ VertexParallelism vertexParallelism, Collection<SlotAssignment> slotAssignments) {
+ this.vertexParallelism = vertexParallelism;
+ this.slotAssignments = slotAssignments;
+ }
+
+ public VertexParallelism getVertexParallelism() {
+ return vertexParallelism;
+ }
+
+ public Collection<SlotAssignment> getSlotAssignments() {
+ return slotAssignments;
+ }
+
+ /**
+ * Create an empty {@link JobSchedulingPlan} with no information about vertices or allocations.
+ */
+ public static JobSchedulingPlan empty() {
+ return new JobSchedulingPlan(VertexParallelism.empty(), Collections.emptyList());
+ }
+
+ /** Assignment of a slot to some target (e.g. a slot sharing group). */
+ public static class SlotAssignment {
+ private final SlotInfo slotInfo;
+ /**
+ * Interpreted by {@link
+ * org.apache.flink.runtime.scheduler.adaptive.allocator.SlotAllocator#tryReserveResources(JobSchedulingPlan)}.
+ * This can be a slot sharing group, a task, or something else.
+ */
+ private final Object target;
+
+ public SlotAssignment(SlotInfo slotInfo, Object target) {
+ this.slotInfo = slotInfo;
+ this.target = target;
+ }
+
+ public SlotInfo getSlotInfo() {
+ return slotInfo;
+ }
+
+ public Object getTarget() {
+ return target;
+ }
+
+ public <T> T getTargetAs(Class<T> clazz) {
+ return (T) getTarget();
+ }
+
+ @Override
+ public String toString() {
+ return String.format(
+ "SlotAssignment: %s, target: %s", slotInfo.getAllocationId(), target);
+ }
+ }
+
+ @Override
+ public String toString() {
+ return String.format(
+ "JobSchedulingPlan: parallelism: %s, assignments: %s",
+ vertexParallelism, slotAssignments);
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotAllocator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotAllocator.java
index 9112b52e876..fea0d659ad3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotAllocator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotAllocator.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.scheduler.adaptive.allocator;
import org.apache.flink.runtime.jobmaster.SlotInfo;
+import org.apache.flink.runtime.scheduler.adaptive.JobSchedulingPlan;
import org.apache.flink.runtime.util.ResourceCounter;
import java.util.Collection;
@@ -43,7 +44,7 @@ public interface SlotAllocator {
* given job information.
*
* <p>Implementations of this method must be side-effect free. There is no guarantee that the
- * result of this method is ever passed to {@link #tryReserveResources(VertexParallelism)}.
+ * result of this method is ever passed to {@link #tryReserveResources(JobSchedulingPlan)}.
*
* @param jobInformation information about the job graph
* @param slots slots to consider for determining the parallelism
@@ -51,7 +52,14 @@ public interface SlotAllocator {
* how the vertices could be assigned to slots, if all vertices could be run with the given
* slots
*/
- Optional<? extends VertexParallelism> determineParallelism(
+ Optional<VertexParallelism> determineParallelism(
+ JobInformation jobInformation, Collection<? extends SlotInfo> slots);
+
+ /**
+ * Same as {@link #determineParallelism(JobInformation, Collection)} but additionally determine
+ * assignment of slots to execution slot sharing groups.
+ */
+ Optional<JobSchedulingPlan> determineParallelismAndCalculateAssignment(
JobInformation jobInformation, Collection<? extends SlotInfo> slots);
/**
@@ -59,9 +67,9 @@ public interface SlotAllocator {
* resources has changed and the reservation with respect to vertexParallelism is no longer
* possible, then this method returns {@link Optional#empty()}.
*
- * @param vertexParallelism information on how slots should be assigned to the slots
+ * @param jobSchedulingPlan information on how slots should be assigned to the slots
* @return Set of reserved slots if the reservation was successful; otherwise {@link
* Optional#empty()}
*/
- Optional<ReservedSlots> tryReserveResources(VertexParallelism vertexParallelism);
+ Optional<ReservedSlots> tryReserveResources(JobSchedulingPlan jobSchedulingPlan);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotSharingSlotAllocator.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotSharingSlotAllocator.java
index ce4c6bfb1af..2b7a9c362ad 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotSharingSlotAllocator.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotSharingSlotAllocator.java
@@ -26,9 +26,10 @@ import org.apache.flink.runtime.jobmaster.LogicalSlot;
import org.apache.flink.runtime.jobmaster.SlotInfo;
import org.apache.flink.runtime.jobmaster.SlotRequestId;
import org.apache.flink.runtime.jobmaster.slotpool.PhysicalSlot;
+import org.apache.flink.runtime.scheduler.adaptive.JobSchedulingPlan;
+import org.apache.flink.runtime.scheduler.adaptive.JobSchedulingPlan.SlotAssignment;
import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
import org.apache.flink.runtime.util.ResourceCounter;
-import org.apache.flink.util.Preconditions;
import javax.annotation.Nonnull;
@@ -93,7 +94,7 @@ public class SlotSharingSlotAllocator implements SlotAllocator {
}
@Override
- public Optional<VertexParallelismWithSlotSharing> determineParallelism(
+ public Optional<VertexParallelism> determineParallelism(
JobInformation jobInformation, Collection<? extends SlotInfo> freeSlots) {
// => less slots than slot-sharing groups
@@ -104,9 +105,6 @@ public class SlotSharingSlotAllocator implements SlotAllocator {
final Map<SlotSharingGroupId, Integer> slotSharingGroupParallelism =
determineSlotsPerSharingGroup(jobInformation, freeSlots.size());
- final Iterator<? extends SlotInfo> slotIterator = freeSlots.iterator();
-
- final Collection<ExecutionSlotSharingGroupAndSlot> assignments = new ArrayList<>();
final Map<JobVertexID, Integer> allVertexParallelism = new HashMap<>();
for (SlotSharingGroup slotSharingGroup : jobInformation.getSlotSharingGroups()) {
@@ -120,21 +118,37 @@ public class SlotSharingSlotAllocator implements SlotAllocator {
containedJobVertices,
slotSharingGroupParallelism.get(
slotSharingGroup.getSlotSharingGroupId()));
+ allVertexParallelism.putAll(vertexParallelism);
+ }
+ return Optional.of(new VertexParallelism(allVertexParallelism));
+ }
- final Iterable<ExecutionSlotSharingGroup> sharedSlotToVertexAssignment =
- createExecutionSlotSharingGroups(vertexParallelism);
-
- for (ExecutionSlotSharingGroup executionSlotSharingGroup :
- sharedSlotToVertexAssignment) {
- final SlotInfo slotInfo = slotIterator.next();
+ @Override
+ public Optional<JobSchedulingPlan> determineParallelismAndCalculateAssignment(
+ JobInformation jobInformation, Collection<? extends SlotInfo> slots) {
+ return determineParallelism(jobInformation, slots)
+ .map(
+ parallelism ->
+ new JobSchedulingPlan(
+ parallelism,
+ assignSlots(jobInformation, slots, parallelism)));
+ }
- assignments.add(
- new ExecutionSlotSharingGroupAndSlot(executionSlotSharingGroup, slotInfo));
- }
- allVertexParallelism.putAll(vertexParallelism);
+ private Collection<SlotAssignment> assignSlots(
+ JobInformation jobInformation,
+ Collection<? extends SlotInfo> freeSlots,
+ VertexParallelism vertexParallelism) {
+ List<ExecutionSlotSharingGroup> allGroups = new ArrayList<>();
+ for (SlotSharingGroup slotSharingGroup : jobInformation.getSlotSharingGroups()) {
+ allGroups.addAll(createExecutionSlotSharingGroups(vertexParallelism, slotSharingGroup));
}
- return Optional.of(new VertexParallelismWithSlotSharing(allVertexParallelism, assignments));
+ Iterator<? extends SlotInfo> iterator = freeSlots.iterator();
+ Collection<SlotAssignment> assignments = new ArrayList<>();
+ for (ExecutionSlotSharingGroup group : allGroups) {
+ assignments.add(new SlotAssignment(iterator.next(), group));
+ }
+ return assignments;
}
/**
@@ -186,15 +200,16 @@ public class SlotSharingSlotAllocator implements SlotAllocator {
return vertexParallelism;
}
- private static Iterable<ExecutionSlotSharingGroup> createExecutionSlotSharingGroups(
- Map<JobVertexID, Integer> containedJobVertices) {
+ private static List<ExecutionSlotSharingGroup> createExecutionSlotSharingGroups(
+ VertexParallelism vertexParallelism, SlotSharingGroup slotSharingGroup) {
final Map<Integer, Set<ExecutionVertexID>> sharedSlotToVertexAssignment = new HashMap<>();
- for (Map.Entry<JobVertexID, Integer> jobVertex : containedJobVertices.entrySet()) {
- for (int i = 0; i < jobVertex.getValue(); i++) {
+ for (JobVertexID jobVertexId : slotSharingGroup.getJobVertexIds()) {
+ int parallelism = vertexParallelism.getParallelism(jobVertexId);
+ for (int subtaskIdx = 0; subtaskIdx < parallelism; subtaskIdx++) {
sharedSlotToVertexAssignment
- .computeIfAbsent(i, ignored -> new HashSet<>())
- .add(new ExecutionVertexID(jobVertex.getKey(), i));
+ .computeIfAbsent(subtaskIdx, ignored -> new HashSet<>())
+ .add(new ExecutionVertexID(jobVertexId, subtaskIdx));
}
}
@@ -204,34 +219,20 @@ public class SlotSharingSlotAllocator implements SlotAllocator {
}
@Override
- public Optional<ReservedSlots> tryReserveResources(VertexParallelism vertexParallelism) {
- Preconditions.checkArgument(
- vertexParallelism instanceof VertexParallelismWithSlotSharing,
- String.format(
- "%s expects %s as argument.",
- SlotSharingSlotAllocator.class.getSimpleName(),
- VertexParallelismWithSlotSharing.class.getSimpleName()));
-
- final VertexParallelismWithSlotSharing vertexParallelismWithSlotSharing =
- (VertexParallelismWithSlotSharing) vertexParallelism;
-
+ public Optional<ReservedSlots> tryReserveResources(JobSchedulingPlan jobSchedulingPlan) {
final Collection<AllocationID> expectedSlots =
- calculateExpectedSlots(vertexParallelismWithSlotSharing.getAssignments());
+ calculateExpectedSlots(jobSchedulingPlan.getSlotAssignments());
if (areAllExpectedSlotsAvailableAndFree(expectedSlots)) {
final Map<ExecutionVertexID, LogicalSlot> assignedSlots = new HashMap<>();
- for (ExecutionSlotSharingGroupAndSlot executionSlotSharingGroup :
- vertexParallelismWithSlotSharing.getAssignments()) {
- final SharedSlot sharedSlot =
- reserveSharedSlot(executionSlotSharingGroup.getSlotInfo());
-
+ for (SlotAssignment assignment : jobSchedulingPlan.getSlotAssignments()) {
+ final SharedSlot sharedSlot = reserveSharedSlot(assignment.getSlotInfo());
for (ExecutionVertexID executionVertexId :
- executionSlotSharingGroup
- .getExecutionSlotSharingGroup()
+ assignment
+ .getTargetAs(ExecutionSlotSharingGroup.class)
.getContainedExecutionVertices()) {
- final LogicalSlot logicalSlot = sharedSlot.allocateLogicalSlot();
- assignedSlots.put(executionVertexId, logicalSlot);
+ assignedSlots.put(executionVertexId, sharedSlot.allocateLogicalSlot());
}
}
@@ -242,11 +243,10 @@ public class SlotSharingSlotAllocator implements SlotAllocator {
}
@Nonnull
- private Collection<AllocationID> calculateExpectedSlots(
- Iterable<? extends ExecutionSlotSharingGroupAndSlot> assignments) {
+ private Collection<AllocationID> calculateExpectedSlots(Iterable<SlotAssignment> assignments) {
final Collection<AllocationID> requiredSlots = new ArrayList<>();
- for (ExecutionSlotSharingGroupAndSlot assignment : assignments) {
+ for (SlotAssignment assignment : assignments) {
requiredSlots.add(assignment.getSlotInfo().getAllocationId());
}
return requiredSlots;
@@ -288,23 +288,4 @@ public class SlotSharingSlotAllocator implements SlotAllocator {
return containedExecutionVertices;
}
}
-
- static class ExecutionSlotSharingGroupAndSlot {
- private final ExecutionSlotSharingGroup executionSlotSharingGroup;
- private final SlotInfo slotInfo;
-
- public ExecutionSlotSharingGroupAndSlot(
- ExecutionSlotSharingGroup executionSlotSharingGroup, SlotInfo slotInfo) {
- this.executionSlotSharingGroup = executionSlotSharingGroup;
- this.slotInfo = slotInfo;
- }
-
- public ExecutionSlotSharingGroup getExecutionSlotSharingGroup() {
- return executionSlotSharingGroup;
- }
-
- public SlotInfo getSlotInfo() {
- return slotInfo;
- }
- }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/VertexParallelism.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/VertexParallelism.java
index 0a19a3217ed..dcc3c10fd18 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/VertexParallelism.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/VertexParallelism.java
@@ -20,17 +20,44 @@ package org.apache.flink.runtime.scheduler.adaptive.allocator;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import java.util.Collection;
+import java.util.Collections;
import java.util.Map;
+import java.util.Optional;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
/**
- * Core result of {@link SlotAllocator#determineParallelism(JobInformation, Collection)}, describing
- * the parallelism each vertex could be scheduled with.
- *
- * <p>{@link SlotAllocator} implementations may encode additional information to be used in {@link
- * SlotAllocator#tryReserveResources(VertexParallelism)}.
+ * Core result of {@link SlotAllocator#determineParallelism(JobInformation, Collection)} among with
+ * {@link org.apache.flink.runtime.scheduler.adaptive.JobSchedulingPlan.SlotAssignment
+ * slotAssignments}, describing the parallelism each vertex could be scheduled with.
*/
-public interface VertexParallelism {
- Map<JobVertexID, Integer> getMaxParallelismForVertices();
+public class VertexParallelism {
+ private final Map<JobVertexID, Integer> parallelismForVertices;
+
+ public VertexParallelism(Map<JobVertexID, Integer> parallelismForVertices) {
+ this.parallelismForVertices = parallelismForVertices;
+ }
+
+ public int getParallelism(JobVertexID jobVertexId) {
+ checkArgument(
+ parallelismForVertices.containsKey(jobVertexId), "Unknown vertex: " + jobVertexId);
+ return parallelismForVertices.get(jobVertexId);
+ }
+
+ public Optional<Integer> getParallelismOptional(JobVertexID jobVertexId) {
+ return Optional.ofNullable(parallelismForVertices.get(jobVertexId));
+ }
+
+ public int getCumulativeParallelism() {
+ return parallelismForVertices.values().stream().reduce(0, Integer::sum);
+ }
+
+ @Override
+ public String toString() {
+ return "VertexParallelism: " + parallelismForVertices;
+ }
- int getParallelism(JobVertexID jobVertexId);
+ public static VertexParallelism empty() {
+ return new VertexParallelism(Collections.emptyMap());
+ }
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/VertexParallelismWithSlotSharing.java b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/VertexParallelismWithSlotSharing.java
deleted file mode 100644
index 05bcadf2728..00000000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/adaptive/allocator/VertexParallelismWithSlotSharing.java
+++ /dev/null
@@ -1,52 +0,0 @@
-/*
- * 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.adaptive.allocator;
-
-import org.apache.flink.runtime.jobgraph.JobVertexID;
-import org.apache.flink.util.Preconditions;
-
-import java.util.Collection;
-import java.util.Map;
-
-/** {@link VertexParallelism} implementation for the {@link SlotSharingSlotAllocator}. */
-public class VertexParallelismWithSlotSharing implements VertexParallelism {
-
- private final Map<JobVertexID, Integer> vertexParallelism;
- private final Collection<SlotSharingSlotAllocator.ExecutionSlotSharingGroupAndSlot> assignments;
-
- VertexParallelismWithSlotSharing(
- Map<JobVertexID, Integer> vertexParallelism,
- Collection<SlotSharingSlotAllocator.ExecutionSlotSharingGroupAndSlot> assignments) {
- this.vertexParallelism = vertexParallelism;
- this.assignments = Preconditions.checkNotNull(assignments);
- }
-
- Iterable<SlotSharingSlotAllocator.ExecutionSlotSharingGroupAndSlot> getAssignments() {
- return assignments;
- }
-
- @Override
- public Map<JobVertexID, Integer> getMaxParallelismForVertices() {
- return vertexParallelism;
- }
-
- @Override
- public int getParallelism(JobVertexID jobVertexId) {
- return vertexParallelism.get(jobVertexId);
- }
-}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java
index e6d457a8ad7..978750ecf3f 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/AdaptiveSchedulerTest.java
@@ -1408,8 +1408,7 @@ public class AdaptiveSchedulerTest extends TestLogger {
final CreatingExecutionGraph.AssignmentResult assignmentResult =
adaptiveScheduler.tryToAssignSlots(
CreatingExecutionGraph.ExecutionGraphWithVertexParallelism.create(
- new StateTrackingMockExecutionGraph(),
- new CreatingExecutionGraphTest.TestingVertexParallelism()));
+ new StateTrackingMockExecutionGraph(), JobSchedulingPlan.empty()));
assertThat(assignmentResult.isSuccess()).isFalse();
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraphTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraphTest.java
index 85a4fe4f3ab..54ebdc11702 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraphTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/CreatingExecutionGraphTest.java
@@ -25,13 +25,11 @@ import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutorServiceAdapter;
import org.apache.flink.runtime.executiongraph.ArchivedExecutionGraph;
import org.apache.flink.runtime.executiongraph.ExecutionGraph;
-import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.metrics.groups.JobManagerJobMetricGroup;
import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
import org.apache.flink.runtime.scheduler.ExecutionGraphHandler;
import org.apache.flink.runtime.scheduler.GlobalFailureHandler;
import org.apache.flink.runtime.scheduler.OperatorCoordinatorHandler;
-import org.apache.flink.runtime.scheduler.adaptive.allocator.VertexParallelism;
import org.apache.flink.runtime.scheduler.exceptionhistory.ExceptionHistoryEntry;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.TestLogger;
@@ -45,7 +43,6 @@ import javax.annotation.Nullable;
import java.time.Duration;
import java.util.List;
-import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledFuture;
@@ -158,8 +155,7 @@ public class CreatingExecutionGraphTest extends TestLogger {
context.setExpectWaitingForResources();
executionGraphWithVertexParallelismFuture.complete(
- CreatingExecutionGraph.ExecutionGraphWithVertexParallelism.create(
- new StateTrackingMockExecutionGraph(), new TestingVertexParallelism()));
+ getGraph(new StateTrackingMockExecutionGraph()));
}
}
@@ -183,9 +179,7 @@ public class CreatingExecutionGraphTest extends TestLogger {
actualExecutionGraph ->
assertThat(actualExecutionGraph).isEqualTo(executionGraph));
- executionGraphWithVertexParallelismFuture.complete(
- CreatingExecutionGraph.ExecutionGraphWithVertexParallelism.create(
- executionGraph, new TestingVertexParallelism()));
+ executionGraphWithVertexParallelismFuture.complete(getGraph(executionGraph));
}
}
@@ -214,9 +208,7 @@ public class CreatingExecutionGraphTest extends TestLogger {
actualExecutionGraph ->
assertThat(actualExecutionGraph).isEqualTo(executionGraph));
- executionGraphWithVertexParallelismFuture.complete(
- CreatingExecutionGraph.ExecutionGraphWithVertexParallelism.create(
- executionGraph, new TestingVertexParallelism()));
+ executionGraphWithVertexParallelismFuture.complete(getGraph(executionGraph));
assertThat(operatorCoordinatorGlobalFailureHandlerRef.get()).isSameAs(context);
}
@@ -353,16 +345,9 @@ public class CreatingExecutionGraphTest extends TestLogger {
}
}
- static final class TestingVertexParallelism implements VertexParallelism {
-
- @Override
- public Map<JobVertexID, Integer> getMaxParallelismForVertices() {
- throw new UnsupportedOperationException("Is not supported");
- }
-
- @Override
- public int getParallelism(JobVertexID jobVertexId) {
- throw new UnsupportedOperationException("Is not supported");
- }
+ private static CreatingExecutionGraph.ExecutionGraphWithVertexParallelism getGraph(
+ StateTrackingMockExecutionGraph executionGraph) {
+ return CreatingExecutionGraph.ExecutionGraphWithVertexParallelism.create(
+ executionGraph, JobSchedulingPlan.empty());
}
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotSharingSlotAllocatorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotSharingSlotAllocatorTest.java
index 8be500919f1..7643acc7410 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotSharingSlotAllocatorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/SlotSharingSlotAllocatorTest.java
@@ -23,6 +23,8 @@ import org.apache.flink.runtime.jobmanager.scheduler.SlotSharingGroup;
import org.apache.flink.runtime.jobmaster.LogicalSlot;
import org.apache.flink.runtime.jobmaster.SlotInfo;
import org.apache.flink.runtime.scheduler.TestingPhysicalSlot;
+import org.apache.flink.runtime.scheduler.adaptive.JobSchedulingPlan;
+import org.apache.flink.runtime.scheduler.adaptive.allocator.SlotSharingSlotAllocator.ExecutionSlotSharingGroup;
import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
import org.apache.flink.runtime.util.ResourceCounter;
import org.apache.flink.util.TestLogger;
@@ -96,15 +98,12 @@ public class SlotSharingSlotAllocatorTest extends TestLogger {
final JobInformation jobInformation =
new TestJobInformation(Arrays.asList(vertex1, vertex2, vertex3));
- final VertexParallelism slotSharingAssignments =
+ final VertexParallelism vertexParallelism =
slotAllocator.determineParallelism(jobInformation, getSlots(2)).get();
- final Map<JobVertexID, Integer> maxParallelismForVertices =
- slotSharingAssignments.getMaxParallelismForVertices();
-
- assertThat(maxParallelismForVertices.get(vertex1.getJobVertexID()), is(1));
- assertThat(maxParallelismForVertices.get(vertex2.getJobVertexID()), is(1));
- assertThat(maxParallelismForVertices.get(vertex3.getJobVertexID()), is(1));
+ assertThat(vertexParallelism.getParallelism(vertex1.getJobVertexID()), is(1));
+ assertThat(vertexParallelism.getParallelism(vertex2.getJobVertexID()), is(1));
+ assertThat(vertexParallelism.getParallelism(vertex3.getJobVertexID()), is(1));
}
@Test
@@ -118,20 +117,17 @@ public class SlotSharingSlotAllocatorTest extends TestLogger {
final JobInformation jobInformation =
new TestJobInformation(Arrays.asList(vertex1, vertex2, vertex3));
- final VertexParallelism slotSharingAssignments =
+ final VertexParallelism vertexParallelism =
slotAllocator.determineParallelism(jobInformation, getSlots(50)).get();
- final Map<JobVertexID, Integer> maxParallelismForVertices =
- slotSharingAssignments.getMaxParallelismForVertices();
-
assertThat(
- maxParallelismForVertices.get(vertex1.getJobVertexID()),
+ vertexParallelism.getParallelism(vertex1.getJobVertexID()),
is(vertex1.getParallelism()));
assertThat(
- maxParallelismForVertices.get(vertex2.getJobVertexID()),
+ vertexParallelism.getParallelism(vertex2.getJobVertexID()),
is(vertex2.getParallelism()));
assertThat(
- maxParallelismForVertices.get(vertex3.getJobVertexID()),
+ vertexParallelism.getParallelism(vertex3.getJobVertexID()),
is(vertex3.getParallelism()));
}
@@ -153,19 +149,18 @@ public class SlotSharingSlotAllocatorTest extends TestLogger {
TestJobInformation testJobInformation =
new TestJobInformation(Arrays.asList(vertex11, vertex12, vertex2));
- Map<JobVertexID, Integer> maxParallelismForVertices =
+ VertexParallelism vertexParallelism =
slotAllocator
.determineParallelism(
testJobInformation,
getSlots(vertex11.getParallelism() + vertex2.getParallelism()))
- .get()
- .getMaxParallelismForVertices();
+ .get();
- Assertions.assertThat(maxParallelismForVertices.get(vertex11.getJobVertexID()))
+ Assertions.assertThat(vertexParallelism.getParallelism(vertex11.getJobVertexID()))
.isEqualTo(vertex11.getParallelism());
- Assertions.assertThat(maxParallelismForVertices.get(vertex12.getJobVertexID()))
+ Assertions.assertThat(vertexParallelism.getParallelism(vertex12.getJobVertexID()))
.isEqualTo(vertex12.getParallelism());
- Assertions.assertThat(maxParallelismForVertices.get(vertex2.getJobVertexID()))
+ Assertions.assertThat(vertexParallelism.getParallelism(vertex2.getJobVertexID()))
.isEqualTo(vertex2.getParallelism());
}
@@ -180,10 +175,10 @@ public class SlotSharingSlotAllocatorTest extends TestLogger {
final JobInformation jobInformation =
new TestJobInformation(Arrays.asList(vertex1, vertex2, vertex3));
- final Optional<VertexParallelismWithSlotSharing> slotSharingAssignments =
+ final Optional<VertexParallelism> vertexParallelism =
slotAllocator.determineParallelism(jobInformation, getSlots(1));
- assertThat(slotSharingAssignments.isPresent(), is(false));
+ assertThat(vertexParallelism.isPresent(), is(false));
}
@Test
@@ -197,20 +192,23 @@ public class SlotSharingSlotAllocatorTest extends TestLogger {
final JobInformation jobInformation =
new TestJobInformation(Arrays.asList(vertex1, vertex2, vertex3));
- final VertexParallelismWithSlotSharing slotAssignments =
- slotAllocator.determineParallelism(jobInformation, getSlots(50)).get();
+ final JobSchedulingPlan jobSchedulingPlan =
+ slotAllocator
+ .determineParallelismAndCalculateAssignment(jobInformation, getSlots(50))
+ .get();
final ReservedSlots reservedSlots =
slotAllocator
- .tryReserveResources(slotAssignments)
+ .tryReserveResources(jobSchedulingPlan)
.orElseThrow(
() -> new RuntimeException("Expected that reservation succeeds."));
final Map<ExecutionVertexID, SlotInfo> expectedAssignments = new HashMap<>();
- for (SlotSharingSlotAllocator.ExecutionSlotSharingGroupAndSlot assignment :
- slotAssignments.getAssignments()) {
+ for (JobSchedulingPlan.SlotAssignment assignment : jobSchedulingPlan.getSlotAssignments()) {
+ ExecutionSlotSharingGroup target =
+ assignment.getTargetAs(ExecutionSlotSharingGroup.class);
for (ExecutionVertexID containedExecutionVertex :
- assignment.getExecutionSlotSharingGroup().getContainedExecutionVertices()) {
+ target.getContainedExecutionVertices()) {
expectedAssignments.put(containedExecutionVertex, assignment.getSlotInfo());
}
}
@@ -234,11 +232,13 @@ public class SlotSharingSlotAllocatorTest extends TestLogger {
final JobInformation jobInformation =
new TestJobInformation(Arrays.asList(vertex1, vertex2, vertex3));
- final VertexParallelismWithSlotSharing slotAssignments =
- slotSharingSlotAllocator.determineParallelism(jobInformation, getSlots(50)).get();
+ JobSchedulingPlan jobSchedulingPlan =
+ slotSharingSlotAllocator
+ .determineParallelismAndCalculateAssignment(jobInformation, getSlots(50))
+ .get();
final Optional<? extends ReservedSlots> reservedSlots =
- slotSharingSlotAllocator.tryReserveResources(slotAssignments);
+ slotSharingSlotAllocator.tryReserveResources(jobSchedulingPlan);
assertFalse(reservedSlots.isPresent());
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/TestingSlotAllocator.java b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/TestingSlotAllocator.java
index aa6fd7bad8d..44b73122ae9 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/TestingSlotAllocator.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/scheduler/adaptive/allocator/TestingSlotAllocator.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.scheduler.adaptive.allocator;
import org.apache.flink.runtime.jobmaster.SlotInfo;
+import org.apache.flink.runtime.scheduler.adaptive.JobSchedulingPlan;
import org.apache.flink.runtime.util.ResourceCounter;
import java.util.Collection;
@@ -33,9 +34,7 @@ public class TestingSlotAllocator implements SlotAllocator {
calculateRequiredSlotsFunction;
private final BiFunction<
- JobInformation,
- Collection<? extends SlotInfo>,
- Optional<? extends VertexParallelism>>
+ JobInformation, Collection<? extends SlotInfo>, Optional<VertexParallelism>>
determineParallelismFunction;
private final Function<VertexParallelism, Optional<ReservedSlots>> tryReserveResourcesFunction;
@@ -43,10 +42,7 @@ public class TestingSlotAllocator implements SlotAllocator {
private TestingSlotAllocator(
Function<Iterable<JobInformation.VertexInformation>, ResourceCounter>
calculateRequiredSlotsFunction,
- BiFunction<
- JobInformation,
- Collection<? extends SlotInfo>,
- Optional<? extends VertexParallelism>>
+ BiFunction<JobInformation, Collection<? extends SlotInfo>, Optional<VertexParallelism>>
determineParallelismFunction,
Function<VertexParallelism, Optional<ReservedSlots>> tryReserveResourcesFunction) {
this.calculateRequiredSlotsFunction = calculateRequiredSlotsFunction;
@@ -61,14 +57,20 @@ public class TestingSlotAllocator implements SlotAllocator {
}
@Override
- public Optional<? extends VertexParallelism> determineParallelism(
+ public Optional<VertexParallelism> determineParallelism(
JobInformation jobInformation, Collection<? extends SlotInfo> slots) {
return determineParallelismFunction.apply(jobInformation, slots);
}
@Override
- public Optional<ReservedSlots> tryReserveResources(VertexParallelism vertexParallelism) {
- return tryReserveResourcesFunction.apply(vertexParallelism);
+ public Optional<JobSchedulingPlan> determineParallelismAndCalculateAssignment(
+ JobInformation jobInformation, Collection<? extends SlotInfo> slots) {
+ return Optional.empty();
+ }
+
+ @Override
+ public Optional<ReservedSlots> tryReserveResources(JobSchedulingPlan jobSchedulingPlan) {
+ return tryReserveResourcesFunction.apply(jobSchedulingPlan.getVertexParallelism());
}
public static Builder newBuilder() {
@@ -80,9 +82,7 @@ public class TestingSlotAllocator implements SlotAllocator {
private Function<Iterable<JobInformation.VertexInformation>, ResourceCounter>
calculateRequiredSlotsFunction = ignored -> ResourceCounter.empty();
private BiFunction<
- JobInformation,
- Collection<? extends SlotInfo>,
- Optional<? extends VertexParallelism>>
+ JobInformation, Collection<? extends SlotInfo>, Optional<VertexParallelism>>
determineParallelismFunction = (ignoredA, ignoredB) -> Optional.empty();
private Function<VertexParallelism, Optional<ReservedSlots>> tryReserveResourcesFunction =
ignored -> Optional.empty();
@@ -98,7 +98,7 @@ public class TestingSlotAllocator implements SlotAllocator {
BiFunction<
JobInformation,
Collection<? extends SlotInfo>,
- Optional<? extends VertexParallelism>>
+ Optional<VertexParallelism>>
determineParallelismFunction) {
this.determineParallelismFunction = determineParallelismFunction;
return this;
[flink] 01/04: [hotfix][tests] Make LocalRecoveryITCase fail when allocations don't match
Posted by ro...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
roman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit d718342ad0311ea7481d1a1bd87c395aeff25928
Author: Roman Khachatryan <kh...@gmail.com>
AuthorDate: Mon Feb 20 20:30:19 2023 +0000
[hotfix][tests] Make LocalRecoveryITCase fail when allocations don't match
Currently, wrong allocation fails the task causing a restart,
which eventually allows to fix the allocation by picking the right TM.
This prevents the test from failure and hides the wrong allocation.
---
.../flink/test/recovery/LocalRecoveryITCase.java | 28 ++++++++++++++++------
1 file changed, 21 insertions(+), 7 deletions(-)
diff --git a/flink-tests/src/test/java/org/apache/flink/test/recovery/LocalRecoveryITCase.java b/flink-tests/src/test/java/org/apache/flink/test/recovery/LocalRecoveryITCase.java
index 1639cda95b9..c8e505c9c9f 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/recovery/LocalRecoveryITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/recovery/LocalRecoveryITCase.java
@@ -19,6 +19,7 @@
package org.apache.flink.test.recovery;
import org.apache.flink.api.common.JobID;
+import org.apache.flink.api.common.accumulators.ListAccumulator;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.CheckpointingOptions;
@@ -67,12 +68,14 @@ import java.util.List;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
-import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertTrue;
/** Tests local recovery by restarting Flink processes. */
@ExtendWith(TestLoggerExtension.class)
class LocalRecoveryITCase {
+ private static final String ALLOCATION_FAILURES_ACCUMULATOR_NAME = "acc";
+
@TempDir private File tmpDirectory;
@Test
@@ -108,7 +111,12 @@ class LocalRecoveryITCase {
restartTaskManagerProcesses(taskManagerProcesses, parallelism - 1);
- jobClient.getJobExecutionResult().get(waitingTimeInSeconds, TimeUnit.SECONDS);
+ List<String> allocFailures =
+ jobClient
+ .getJobExecutionResult()
+ .get(waitingTimeInSeconds, TimeUnit.SECONDS)
+ .getAccumulatorResult(ALLOCATION_FAILURES_ACCUMULATOR_NAME);
+ assertTrue(allocFailures.isEmpty(), allocFailures.toString());
success = true;
} finally {
@@ -307,11 +315,17 @@ class LocalRecoveryITCase {
new IllegalStateException(
"Could not find corresponding TaskNameAllocationID information."));
- assertThat(myTaskNameAllocationId.getAllocationId())
- .withFailMessage(
- "The task was deployed to AllocationID(%s) but it should have been deployed to AllocationID(%s) for local recovery.",
- allocationId, myTaskNameAllocationId.getAllocationId())
- .isEqualTo(allocationId);
+ runtimeContext.addAccumulator(
+ ALLOCATION_FAILURES_ACCUMULATOR_NAME, new ListAccumulator<String>());
+ if (!allocationId.equals(myTaskNameAllocationId.getAllocationId())) {
+ runtimeContext
+ .getAccumulator(ALLOCATION_FAILURES_ACCUMULATOR_NAME)
+ .add(
+ String.format(
+ "The task was deployed to AllocationID(%s) but it should have been deployed to AllocationID(%s) for local recovery.",
+ allocationId,
+ myTaskNameAllocationId.getAllocationId()));
+ }
// terminate
running = false;
}