This is an automated email from the ASF dual-hosted git repository.
chesnay pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 07bab2f72ad [FLINK-27222][coordination] Decouple last (al)location
from execution history
07bab2f72ad is described below
commit 07bab2f72add9ecdffd8ffa038b552797ff72579
Author: Chesnay Schepler <[email protected]>
AuthorDate: Fri Apr 15 09:16:56 2022 +0200
[FLINK-27222][coordination] Decouple last (al)location from execution
history
---
.../flink/runtime/executiongraph/Execution.java | 4 ++
.../runtime/executiongraph/ExecutionVertex.java | 50 ++++++++--------------
.../flink/runtime/scheduler/DefaultScheduler.java | 2 +-
.../executiongraph/ExecutionVertexTest.java | 14 ++++--
4 files changed, 32 insertions(+), 38 deletions(-)
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
index 352a184d62f..487d4e3da77 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/Execution.java
@@ -288,6 +288,10 @@ public class Execution
&& !taskManagerLocationFuture.isDone()) {
taskManagerLocationFuture.complete(logicalSlot.getTaskManagerLocation());
assignedAllocationID = logicalSlot.getAllocationId();
+ getVertex()
+ .setLatestPriorSlotAllocation(
+
assignedResource.getTaskManagerLocation(),
+ logicalSlot.getAllocationId());
return true;
} else {
// free assigned resource and return false
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
index fef96e27f10..4bf20f3faab 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/executiongraph/ExecutionVertex.java
@@ -36,6 +36,7 @@ import
org.apache.flink.runtime.scheduler.strategy.ConsumedPartitionGroup;
import org.apache.flink.runtime.scheduler.strategy.ExecutionVertexID;
import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
import org.apache.flink.runtime.util.EvictingBoundedList;
+import org.apache.flink.util.Preconditions;
import javax.annotation.Nullable;
@@ -47,7 +48,6 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
-import java.util.function.Function;
import static org.apache.flink.runtime.execution.ExecutionState.FINISHED;
import static org.apache.flink.util.Preconditions.checkArgument;
@@ -84,6 +84,11 @@ public class ExecutionVertex
private final ArrayList<InputSplit> inputSplits;
+ /** This field holds the allocation id of the last successful assignment.
*/
+ @Nullable private TaskManagerLocation lastAssignedLocation;
+
+ @Nullable private AllocationID lastAssignedAllocationID;
+
//
--------------------------------------------------------------------------------------------
/**
@@ -280,44 +285,23 @@ public class ExecutionVertex
}
}
- /**
- * Gets the latest property from a prior execution that is not null.
- *
- * @param extractor defining the property to extract
- * @param <T> type of the property
- * @return Optional containing the latest property if it exists; otherwise
{@code
- * Optional.empty()}.
- */
- private <T> Optional<T> getLatestPriorProperty(Function<ArchivedExecution,
T> extractor) {
- int index = priorExecutions.size() - 1;
-
- while (index >= 0 && !priorExecutions.isDroppedIndex(index)) {
- final ArchivedExecution archivedExecution =
priorExecutions.get(index);
-
- final T extractedValue = extractor.apply(archivedExecution);
-
- if (extractedValue != null) {
- return Optional.of(extractedValue);
- }
-
- index -= 1;
- }
-
- return Optional.empty();
+ void setLatestPriorSlotAllocation(
+ TaskManagerLocation taskManagerLocation, AllocationID
lastAssignedAllocationID) {
+ this.lastAssignedLocation =
Preconditions.checkNotNull(taskManagerLocation);
+ this.lastAssignedAllocationID =
Preconditions.checkNotNull(lastAssignedAllocationID);
}
/**
- * Gets the location where the latest completed/canceled/failed execution
of the vertex's task
- * happened.
+ * Gets the location that an execution of this vertex was assigned to.
*
- * @return The latest prior execution location, or null, if there is none,
yet.
+ * @return The last execution location, or null, if there is none, yet.
*/
- public Optional<TaskManagerLocation> findLatestPriorLocation() {
- return
getLatestPriorProperty(ArchivedExecution::getAssignedResourceLocation);
+ public Optional<TaskManagerLocation> findLastLocation() {
+ return Optional.ofNullable(lastAssignedLocation);
}
- public Optional<AllocationID> findLatestPriorAllocation() {
- return
getLatestPriorProperty(ArchivedExecution::getAssignedAllocationID);
+ public Optional<AllocationID> findLastAllocation() {
+ return Optional.ofNullable(lastAssignedAllocationID);
}
EvictingBoundedList<ArchivedExecution> getCopyOfPriorExecutionsList() {
@@ -353,7 +337,7 @@ public class ExecutionVertex
// only restore to same execution if it has state
if (currentExecution.getTaskRestore() != null
&&
currentExecution.getTaskRestore().getTaskStateSnapshot().hasState()) {
- return findLatestPriorLocation();
+ return findLastLocation();
}
return Optional.empty();
diff --git
a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultScheduler.java
b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultScheduler.java
index 3af4b61ece2..e9e369c3d19 100644
---
a/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultScheduler.java
+++
b/flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/DefaultScheduler.java
@@ -732,7 +732,7 @@ public class DefaultScheduler extends SchedulerBase
implements SchedulerOperatio
@Override
public Optional<AllocationID> findPriorAllocationId(
final ExecutionVertexID executionVertexId) {
- return
getExecutionVertex(executionVertexId).findLatestPriorAllocation();
+ return getExecutionVertex(executionVertexId).findLastAllocation();
}
@Override
diff --git
a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexTest.java
b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexTest.java
index 8c18a448406..dd87d04abcb 100644
---
a/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexTest.java
+++
b/flink-runtime/src/test/java/org/apache/flink/runtime/executiongraph/ExecutionVertexTest.java
@@ -18,6 +18,8 @@
package org.apache.flink.runtime.executiongraph;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.configuration.JobManagerOptions;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import
org.apache.flink.runtime.concurrent.ComponentMainThreadExecutorServiceAdapter;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
@@ -116,11 +118,15 @@ public class ExecutionVertexTest extends TestLogger {
final JobGraph jobGraph = JobGraphTestUtils.streamingJobGraph(source);
final TestingPhysicalSlotProvider withLimitedAmountOfPhysicalSlots =
TestingPhysicalSlotProvider.createWithLimitedAmountOfPhysicalSlots(1);
+ final Configuration configuration = new Configuration();
+ // make sure that retrieving the last (al)location is independent from
the history size
+ configuration.set(JobManagerOptions.MAX_ATTEMPTS_HISTORY_SIZE, 1);
final SchedulerBase scheduler =
new SchedulerTestingUtils.DefaultSchedulerBuilder(
jobGraph,
ComponentMainThreadExecutorServiceAdapter.forMainThread(),
EXECUTOR_RESOURCE.getExecutor())
+ .setJobMasterConfiguration(configuration)
.setExecutionSlotAllocatorFactory(
SchedulerTestingUtils.newSlotSharingExecutionSlotAllocatorFactory(
withLimitedAmountOfPhysicalSlots))
@@ -142,15 +148,15 @@ public class ExecutionVertexTest extends TestLogger {
cancelExecution(firstExecution);
sourceExecutionVertex.resetForNewExecution();
-
assertThat(sourceExecutionVertex.findLatestPriorAllocation()).hasValue(allocationId);
-
assertThat(sourceExecutionVertex.findLatestPriorLocation()).hasValue(taskManagerLocation);
+
assertThat(sourceExecutionVertex.findLastAllocation()).hasValue(allocationId);
+
assertThat(sourceExecutionVertex.findLastLocation()).hasValue(taskManagerLocation);
final Execution secondExecution =
sourceExecutionVertex.getCurrentExecutionAttempt();
cancelExecution(secondExecution);
sourceExecutionVertex.resetForNewExecution();
-
assertThat(sourceExecutionVertex.findLatestPriorAllocation()).hasValue(allocationId);
-
assertThat(sourceExecutionVertex.findLatestPriorLocation()).hasValue(taskManagerLocation);
+
assertThat(sourceExecutionVertex.findLastAllocation()).hasValue(allocationId);
+
assertThat(sourceExecutionVertex.findLastLocation()).hasValue(taskManagerLocation);
}
private void cancelExecution(Execution execution) {
