You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by gu...@apache.org on 2022/04/27 03:41:13 UTC
[flink] branch release-1.14 updated: [FLINK-27409][runtime] Cleanup stale slot allocation record when the resource requirement of a job is empty
This is an automated email from the ASF dual-hosted git repository.
guoyangze pushed a commit to branch release-1.14
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.14 by this push:
new b5d77c0cdb1 [FLINK-27409][runtime] Cleanup stale slot allocation record when the resource requirement of a job is empty
b5d77c0cdb1 is described below
commit b5d77c0cdb1519163acad92084a4f3f79b24b012
Author: Yangze Guo <ka...@gmail.com>
AuthorDate: Tue Apr 26 14:34:35 2022 +0800
[FLINK-27409][runtime] Cleanup stale slot allocation record when the resource requirement of a job is empty
This closes #19582.
---
.../slotmanager/FineGrainedSlotManager.java | 14 +--
.../slotmanager/FineGrainedTaskManagerTracker.java | 6 ++
.../slotmanager/TaskManagerTracker.java | 7 ++
.../AbstractFineGrainedSlotManagerITCase.java | 107 +++++++++++++++++++++
4 files changed, 128 insertions(+), 6 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManager.java
index 8872d8c3bab..c6a136a8858 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedSlotManager.java
@@ -255,6 +255,7 @@ public class FineGrainedSlotManager implements SlotManager {
@Override
public void clearResourceRequirements(JobID jobId) {
jobMasterTargetAddresses.remove(jobId);
+ taskManagerTracker.clearPendingAllocationsOfJob(jobId);
resourceTracker.notifyResourceRequirements(jobId, Collections.emptyList());
}
@@ -263,22 +264,23 @@ public class FineGrainedSlotManager implements SlotManager {
checkInit();
if (resourceRequirements.getResourceRequirements().isEmpty()
&& resourceTracker.isRequirementEmpty(resourceRequirements.getJobId())) {
+ // Skip duplicate empty resource requirements.
return;
- } else if (resourceRequirements.getResourceRequirements().isEmpty()) {
+ }
+
+ if (resourceRequirements.getResourceRequirements().isEmpty()) {
LOG.info("Clearing resource requirements of job {}", resourceRequirements.getJobId());
+ jobMasterTargetAddresses.remove(resourceRequirements.getJobId());
+ taskManagerTracker.clearPendingAllocationsOfJob(resourceRequirements.getJobId());
} else {
LOG.info(
"Received resource requirements from job {}: {}",
resourceRequirements.getJobId(),
resourceRequirements.getResourceRequirements());
- }
-
- if (resourceRequirements.getResourceRequirements().isEmpty()) {
- jobMasterTargetAddresses.remove(resourceRequirements.getJobId());
- } else {
jobMasterTargetAddresses.put(
resourceRequirements.getJobId(), resourceRequirements.getTargetAddress());
}
+
resourceTracker.notifyResourceRequirements(
resourceRequirements.getJobId(), resourceRequirements.getResourceRequirements());
checkResourceRequirementsWithDelay();
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedTaskManagerTracker.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedTaskManagerTracker.java
index 35dd6dc341b..5bb4eaf5cc9 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedTaskManagerTracker.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/FineGrainedTaskManagerTracker.java
@@ -79,6 +79,12 @@ public class FineGrainedTaskManagerTracker implements TaskManagerTracker {
pendingSlotAllocationRecords.putAll(pendingSlotAllocations);
}
+ @Override
+ public void clearPendingAllocationsOfJob(JobID jobId) {
+ LOG.info("Clear all pending allocations for job {}.", jobId);
+ pendingSlotAllocationRecords.values().forEach(allocation -> allocation.remove(jobId));
+ }
+
@Override
public void addTaskManager(
TaskExecutorConnection taskExecutorConnection,
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/TaskManagerTracker.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/TaskManagerTracker.java
index 88852c22643..ee8e3d14467 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/TaskManagerTracker.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/TaskManagerTracker.java
@@ -97,6 +97,13 @@ interface TaskManagerTracker
void replaceAllPendingAllocations(
Map<PendingTaskManagerId, Map<JobID, ResourceCounter>> pendingSlotAllocations);
+ /**
+ * Clear all previous pending slot allocation records for the given job.
+ *
+ * @param jobId of the given job
+ */
+ void clearPendingAllocationsOfJob(JobID jobId);
+
/** Removes all state from the tracker. */
void clear();
}
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/AbstractFineGrainedSlotManagerITCase.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/AbstractFineGrainedSlotManagerITCase.java
index 6fab2b13b89..466e521905d 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/AbstractFineGrainedSlotManagerITCase.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/AbstractFineGrainedSlotManagerITCase.java
@@ -383,6 +383,113 @@ public abstract class AbstractFineGrainedSlotManagerITCase extends FineGrainedSl
};
}
+ @Test
+ public void testRegisterPendingResourceAfterClearingRequirement() throws Exception {
+ final CompletableFuture<AllocationID> allocationIdFuture = new CompletableFuture<>();
+ final CompletableFuture<Void> allocateResourceFutures = new CompletableFuture<>();
+ final CompletableFuture<Void> registerFuture = new CompletableFuture<>();
+ final ResourceRequirements resourceRequirements = createResourceRequirementsForSingleSlot();
+ final TaskExecutorGateway taskExecutorGateway =
+ new TestingTaskExecutorGatewayBuilder()
+ .setRequestSlotFunction(
+ tuple6 -> {
+ allocationIdFuture.complete(tuple6.f2);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ })
+ .createTestingTaskExecutorGateway();
+ final ResourceID resourceID = ResourceID.generate();
+ final TaskExecutorConnection taskManagerConnection =
+ new TaskExecutorConnection(resourceID, taskExecutorGateway);
+ final SlotReport slotReport = new SlotReport();
+ new Context() {
+ {
+ resourceActionsBuilder.setAllocateResourceConsumer(
+ ignored -> allocateResourceFutures.complete(null));
+ runTest(
+ () -> {
+ runInMainThread(
+ () ->
+ getSlotManager()
+ .processResourceRequirements(
+ resourceRequirements));
+ assertFutureCompleteAndReturn(allocateResourceFutures);
+ runInMainThread(
+ () -> {
+ getSlotManager()
+ .clearResourceRequirements(
+ resourceRequirements.getJobId());
+ getSlotManager()
+ .registerTaskManager(
+ taskManagerConnection,
+ slotReport,
+ DEFAULT_TOTAL_RESOURCE_PROFILE,
+ DEFAULT_SLOT_RESOURCE_PROFILE);
+ registerFuture.complete(null);
+ });
+ assertFutureCompleteAndReturn(registerFuture);
+ assertFutureNotComplete(allocationIdFuture);
+ assertEquals(
+ getTaskManagerTracker().getPendingTaskManagers().size(), 0);
+ });
+ }
+ };
+ }
+
+ @Test
+ public void testRegisterPendingResourceAfterEmptyResourceRequirement() throws Exception {
+ final CompletableFuture<AllocationID> allocationIdFuture = new CompletableFuture<>();
+ final CompletableFuture<Void> allocateResourceFutures = new CompletableFuture<>();
+ final CompletableFuture<Void> registerFuture = new CompletableFuture<>();
+ final ResourceRequirements resourceRequirements = createResourceRequirementsForSingleSlot();
+ final TaskExecutorGateway taskExecutorGateway =
+ new TestingTaskExecutorGatewayBuilder()
+ .setRequestSlotFunction(
+ tuple6 -> {
+ allocationIdFuture.complete(tuple6.f2);
+ return CompletableFuture.completedFuture(Acknowledge.get());
+ })
+ .createTestingTaskExecutorGateway();
+ final ResourceID resourceID = ResourceID.generate();
+ final TaskExecutorConnection taskManagerConnection =
+ new TaskExecutorConnection(resourceID, taskExecutorGateway);
+ final SlotReport slotReport = new SlotReport();
+ new Context() {
+ {
+ resourceActionsBuilder.setAllocateResourceConsumer(
+ ignored -> allocateResourceFutures.complete(null));
+ runTest(
+ () -> {
+ runInMainThread(
+ () ->
+ getSlotManager()
+ .processResourceRequirements(
+ resourceRequirements));
+ assertFutureCompleteAndReturn(allocateResourceFutures);
+ runInMainThread(
+ () -> {
+ getSlotManager()
+ .processResourceRequirements(
+ ResourceRequirements.empty(
+ resourceRequirements.getJobId(),
+ resourceRequirements
+ .getTargetAddress()));
+ getSlotManager()
+ .registerTaskManager(
+ taskManagerConnection,
+ slotReport,
+ DEFAULT_TOTAL_RESOURCE_PROFILE,
+ DEFAULT_SLOT_RESOURCE_PROFILE);
+ registerFuture.complete(null);
+ });
+ assertFutureCompleteAndReturn(registerFuture);
+ assertFutureNotComplete(allocationIdFuture);
+ assertEquals(
+ getTaskManagerTracker().getPendingTaskManagers().size(), 0);
+ });
+ }
+ };
+ }
+
/**
* Tests that we only request new resources/containers once we have assigned all pending task
* managers.