You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ch...@apache.org on 2021/04/15 00:20:08 UTC
[flink] 02/02: [FLINK-22180][coordination] Only reclaim slots if
job is removed
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
commit 9fd6ecf18ddb744a971f575ab11eb19b44383899
Author: Chesnay Schepler <ch...@apache.org>
AuthorDate: Wed Apr 14 11:15:40 2021 +0200
[FLINK-22180][coordination] Only reclaim slots if job is removed
---
.../runtime/resourcemanager/ResourceManager.java | 3 +-
.../slotmanager/DeclarativeSlotManager.java | 14 +++--
.../slotmanager/FineGrainedSlotManager.java | 6 ++
.../resourcemanager/slotmanager/SlotManager.java | 9 +++
.../slotmanager/SlotManagerImpl.java | 3 +
.../resourcemanager/ResourceManagerTest.java | 67 ++++++++++++++++++++++
.../slotmanager/DeclarativeSlotManagerTest.java | 45 +++++++++++++--
.../slotmanager/TestingSlotManager.java | 18 +++++-
.../slotmanager/TestingSlotManagerBuilder.java | 21 ++++++-
9 files changed, 171 insertions(+), 15 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
index 04e8fff..421ca45 100755
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
@@ -1055,8 +1055,7 @@ public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable>
jmResourceIdRegistrations.remove(jobManagerResourceId);
if (resourceRequirementHandling == ResourceRequirementHandling.CLEAR) {
- slotManager.processResourceRequirements(
- ResourceRequirements.empty(jobId, jobMasterGateway.getAddress()));
+ slotManager.clearResourceRequirements(jobId);
}
// tell the job manager about the disconnect
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/DeclarativeSlotManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/DeclarativeSlotManager.java
index 7f1d724..1cc4337 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/DeclarativeSlotManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/DeclarativeSlotManager.java
@@ -246,6 +246,14 @@ public class DeclarativeSlotManager implements SlotManager {
// ---------------------------------------------------------------------------------------------
@Override
+ public void clearResourceRequirements(JobID jobId) {
+ checkInit();
+ maybeReclaimInactiveSlots(jobId);
+ jobMasterTargetAddresses.remove(jobId);
+ resourceTracker.notifyResourceRequirements(jobId, Collections.emptyList());
+ }
+
+ @Override
public void processResourceRequirements(ResourceRequirements resourceRequirements) {
checkInit();
if (resourceRequirements.getResourceRequirements().isEmpty()) {
@@ -257,11 +265,7 @@ public class DeclarativeSlotManager implements SlotManager {
resourceRequirements.getResourceRequirements());
}
- if (resourceRequirements.getResourceRequirements().isEmpty()) {
- jobMasterTargetAddresses.remove(resourceRequirements.getJobId());
-
- maybeReclaimInactiveSlots(resourceRequirements.getJobId());
- } else {
+ if (!resourceRequirements.getResourceRequirements().isEmpty()) {
jobMasterTargetAddresses.put(
resourceRequirements.getJobId(), resourceRequirements.getTargetAddress());
}
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 27dd15b..a775c93 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
@@ -257,6 +257,12 @@ public class FineGrainedSlotManager implements SlotManager {
// ---------------------------------------------------------------------------------------------
@Override
+ public void clearResourceRequirements(JobID jobId) {
+ jobMasterTargetAddresses.remove(jobId);
+ resourceTracker.notifyResourceRequirements(jobId, Collections.emptyList());
+ }
+
+ @Override
public void processResourceRequirements(ResourceRequirements resourceRequirements) {
checkInit();
LOG.debug(
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManager.java
index ba19cbb..52ef64b 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManager.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.resourcemanager.slotmanager;
+import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
import org.apache.flink.runtime.clusterframework.types.SlotID;
@@ -92,6 +93,14 @@ public interface SlotManager extends AutoCloseable {
void suspend();
/**
+ * Notifies the slot manager that the resource requirements for the given job should be cleared.
+ * The slot manager may assume that no further updates to the resource requirements will occur.
+ *
+ * @param jobId job for which to clear the requirements
+ */
+ void clearResourceRequirements(JobID jobId);
+
+ /**
* Notifies the slot manager about the resource requirements of a job.
*
* @param resourceRequirements resource requirements of a job
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerImpl.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerImpl.java
index e4b45da..c17ba46 100755
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerImpl.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerImpl.java
@@ -394,6 +394,9 @@ public class SlotManagerImpl implements SlotManager {
// ---------------------------------------------------------------------------------------------
@Override
+ public void clearResourceRequirements(JobID jobId) {}
+
+ @Override
public void processResourceRequirements(ResourceRequirements resourceRequirements) {
// no-op; don't throw an UnsupportedOperationException here because there are code paths
// where the resource
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerTest.java
index c073a15..0704451 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerTest.java
@@ -37,9 +37,12 @@ import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
import org.apache.flink.runtime.registration.RegistrationResponse;
import org.apache.flink.runtime.resourcemanager.slotmanager.SlotManager;
import org.apache.flink.runtime.resourcemanager.slotmanager.SlotManagerBuilder;
+import org.apache.flink.runtime.resourcemanager.slotmanager.TestingSlotManagerBuilder;
import org.apache.flink.runtime.rest.messages.taskmanager.TaskManagerInfo;
import org.apache.flink.runtime.rpc.RpcUtils;
import org.apache.flink.runtime.rpc.TestingRpcService;
+import org.apache.flink.runtime.slots.ResourceRequirement;
+import org.apache.flink.runtime.slots.ResourceRequirements;
import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
import org.apache.flink.runtime.taskexecutor.TaskExecutorMemoryConfiguration;
import org.apache.flink.runtime.taskexecutor.TaskExecutorThreadInfoGateway;
@@ -56,6 +59,7 @@ import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
+import java.util.Collections;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
@@ -228,6 +232,60 @@ public class ResourceManagerTest extends TestLogger {
}
@Test
+ public void testDisconnectJobManagerClearsRequirements() throws Exception {
+ final TestingJobMasterGateway jobMasterGateway =
+ new TestingJobMasterGatewayBuilder()
+ .setAddress(UUID.randomUUID().toString())
+ .build();
+ rpcService.registerGateway(jobMasterGateway.getAddress(), jobMasterGateway);
+
+ final JobLeaderIdService jobLeaderIdService =
+ TestingJobLeaderIdService.newBuilder()
+ .setGetLeaderIdFunction(
+ jobId ->
+ CompletableFuture.completedFuture(
+ jobMasterGateway.getFencingToken()))
+ .build();
+
+ final CompletableFuture<JobID> clearRequirementsFuture = new CompletableFuture<>();
+
+ final SlotManager slotManager =
+ new TestingSlotManagerBuilder()
+ .setClearRequirementsConsumer(clearRequirementsFuture::complete)
+ .createSlotManager();
+ resourceManager =
+ createAndStartResourceManager(heartbeatServices, jobLeaderIdService, slotManager);
+
+ final JobID jobId = JobID.generate();
+ final ResourceManagerGateway resourceManagerGateway =
+ resourceManager.getSelfGateway(ResourceManagerGateway.class);
+ resourceManagerGateway
+ .registerJobManager(
+ jobMasterGateway.getFencingToken(),
+ ResourceID.generate(),
+ jobMasterGateway.getAddress(),
+ jobId,
+ TIMEOUT)
+ .get();
+
+ resourceManagerGateway
+ .declareRequiredResources(
+ jobMasterGateway.getFencingToken(),
+ ResourceRequirements.create(
+ jobId,
+ jobMasterGateway.getAddress(),
+ Collections.singleton(
+ ResourceRequirement.create(ResourceProfile.UNKNOWN, 1))),
+ TIMEOUT)
+ .get();
+
+ resourceManagerGateway.disconnectJobManager(
+ jobId, JobStatus.FINISHED, new FlinkException("Test exception"));
+
+ assertThat(clearRequirementsFuture.get(5, TimeUnit.SECONDS), is(jobId));
+ }
+
+ @Test
public void testHeartbeatTimeoutWithJobMaster() throws Exception {
final CompletableFuture<ResourceID> heartbeatRequestFuture = new CompletableFuture<>();
final CompletableFuture<ResourceManagerId> disconnectFuture = new CompletableFuture<>();
@@ -396,6 +454,15 @@ public class ResourceManagerTest extends TestLogger {
.setScheduledExecutor(rpcService.getScheduledExecutor())
.build();
+ return createAndStartResourceManager(heartbeatServices, jobLeaderIdService, slotManager);
+ }
+
+ private TestingResourceManager createAndStartResourceManager(
+ HeartbeatServices heartbeatServices,
+ JobLeaderIdService jobLeaderIdService,
+ SlotManager slotManager)
+ throws Exception {
+
final TestingResourceManager resourceManager =
new TestingResourceManager(
rpcService,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/DeclarativeSlotManagerTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/DeclarativeSlotManagerTest.java
index 7c9d24a..88a130a 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/DeclarativeSlotManagerTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/DeclarativeSlotManagerTest.java
@@ -1375,8 +1375,7 @@ public class DeclarativeSlotManagerTest extends TestLogger {
}
@Test
- public void testReclaimInactiveSlotsOnEmptyRequirements() throws Exception {
-
+ public void testReclaimInactiveSlotsOnClearRequirements() throws Exception {
final CompletableFuture<JobID> freeInactiveSlotsJobIdFuture = new CompletableFuture<>();
final TestingTaskExecutorGateway taskExecutorGateway =
@@ -1405,16 +1404,52 @@ public class DeclarativeSlotManagerTest extends TestLogger {
slotManager.processResourceRequirements(createResourceRequirements(jobId, 2));
assertThat(freeInactiveSlotsJobIdFuture.isDone(), is(false));
- // decrease requirements, which should not trigger slots being reclaimed
- slotManager.processResourceRequirements(createResourceRequirements(jobId, 1));
+ // set requirements to 0, which should not trigger slots being reclaimed
+ slotManager.processResourceRequirements(ResourceRequirements.empty(jobId, "foobar"));
assertThat(freeInactiveSlotsJobIdFuture.isDone(), is(false));
// clear requirements, which should trigger slots being reclaimed
- slotManager.processResourceRequirements(ResourceRequirements.empty(jobId, "foobar"));
+ slotManager.clearResourceRequirements(jobId);
assertThat(freeInactiveSlotsJobIdFuture.get(), is(jobId));
}
}
+ @Test
+ public void testClearRequirementsClearsResourceTracker() throws Exception {
+ final ResourceTracker resourceTracker = new DefaultResourceTracker();
+
+ final CompletableFuture<JobID> freeInactiveSlotsJobIdFuture = new CompletableFuture<>();
+
+ final TestingTaskExecutorGateway taskExecutorGateway =
+ new TestingTaskExecutorGatewayBuilder()
+ .setFreeInactiveSlotsConsumer(freeInactiveSlotsJobIdFuture::complete)
+ .createTestingTaskExecutorGateway();
+
+ try (final DeclarativeSlotManager slotManager =
+ createDeclarativeSlotManagerBuilder()
+ .setResourceTracker(resourceTracker)
+ .buildAndStart(
+ ResourceManagerId.generate(),
+ ComponentMainThreadExecutorServiceAdapter.forMainThread(),
+ new TestingResourceActionsBuilder().build())) {
+
+ final JobID jobId = new JobID();
+
+ final TaskExecutorConnection taskExecutionConnection =
+ createTaskExecutorConnection(taskExecutorGateway);
+ final SlotReport slotReport =
+ createSlotReportWithAllocatedSlots(
+ taskExecutionConnection.getResourceID(), jobId, 1);
+ slotManager.registerTaskManager(
+ taskExecutionConnection, slotReport, ResourceProfile.ANY, ResourceProfile.ANY);
+
+ slotManager.processResourceRequirements(createResourceRequirements(jobId, 2));
+ slotManager.clearResourceRequirements(jobId);
+
+ assertThat(resourceTracker.getMissingResources().keySet(), empty());
+ }
+ }
+
private static SlotReport createSlotReport(ResourceID taskExecutorResourceId, int numberSlots) {
final Set<SlotStatus> slotStatusSet = new HashSet<>(numberSlots);
for (int i = 0; i < numberSlots; i++) {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/TestingSlotManager.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/TestingSlotManager.java
index d1fe0eb..0bc44fd 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/TestingSlotManager.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/TestingSlotManager.java
@@ -18,6 +18,7 @@
package org.apache.flink.runtime.resourcemanager.slotmanager;
+import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
import org.apache.flink.runtime.clusterframework.types.SlotID;
@@ -42,12 +43,18 @@ public class TestingSlotManager implements SlotManager {
private final Consumer<Boolean> setFailUnfulfillableRequestConsumer;
private final Supplier<Map<WorkerResourceSpec, Integer>> getRequiredResourcesSupplier;
+ private final Consumer<ResourceRequirements> processRequirementsConsumer;
+ private final Consumer<JobID> clearRequirementsConsumer;
TestingSlotManager(
Consumer<Boolean> setFailUnfulfillableRequestConsumer,
- Supplier<Map<WorkerResourceSpec, Integer>> getRequiredResourcesSupplier) {
+ Supplier<Map<WorkerResourceSpec, Integer>> getRequiredResourcesSupplier,
+ Consumer<ResourceRequirements> processRequirementsConsumer,
+ Consumer<JobID> clearRequirementsConsumer) {
this.setFailUnfulfillableRequestConsumer = setFailUnfulfillableRequestConsumer;
this.getRequiredResourcesSupplier = getRequiredResourcesSupplier;
+ this.processRequirementsConsumer = processRequirementsConsumer;
+ this.clearRequirementsConsumer = clearRequirementsConsumer;
}
@Override
@@ -115,7 +122,14 @@ public class TestingSlotManager implements SlotManager {
public void suspend() {}
@Override
- public void processResourceRequirements(ResourceRequirements resourceRequirements) {}
+ public void clearResourceRequirements(JobID jobId) {
+ clearRequirementsConsumer.accept(jobId);
+ }
+
+ @Override
+ public void processResourceRequirements(ResourceRequirements resourceRequirements) {
+ processRequirementsConsumer.accept(resourceRequirements);
+ }
@Override
public boolean registerSlotRequest(SlotRequest slotRequest) {
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/TestingSlotManagerBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/TestingSlotManagerBuilder.java
index 9243d77..d89c6dc 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/TestingSlotManagerBuilder.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/slotmanager/TestingSlotManagerBuilder.java
@@ -18,7 +18,9 @@
package org.apache.flink.runtime.resourcemanager.slotmanager;
+import org.apache.flink.api.common.JobID;
import org.apache.flink.runtime.resourcemanager.WorkerResourceSpec;
+import org.apache.flink.runtime.slots.ResourceRequirements;
import java.util.Collections;
import java.util.Map;
@@ -31,6 +33,8 @@ public class TestingSlotManagerBuilder {
private Consumer<Boolean> setFailUnfulfillableRequestConsumer = ignored -> {};
private Supplier<Map<WorkerResourceSpec, Integer>> getRequiredResourcesSupplier =
() -> Collections.emptyMap();
+ private Consumer<ResourceRequirements> processRequirementsConsumer = ignored -> {};
+ private Consumer<JobID> clearRequirementsConsumer = ignored -> {};
public TestingSlotManagerBuilder setSetFailUnfulfillableRequestConsumer(
Consumer<Boolean> setFailUnfulfillableRequestConsumer) {
@@ -44,8 +48,23 @@ public class TestingSlotManagerBuilder {
return this;
}
+ public TestingSlotManagerBuilder setProcessRequirementsConsumer(
+ Consumer<ResourceRequirements> processRequirementsConsumer) {
+ this.processRequirementsConsumer = processRequirementsConsumer;
+ return this;
+ }
+
+ public TestingSlotManagerBuilder setClearRequirementsConsumer(
+ Consumer<JobID> clearRequirementsConsumer) {
+ this.clearRequirementsConsumer = clearRequirementsConsumer;
+ return this;
+ }
+
public TestingSlotManager createSlotManager() {
return new TestingSlotManager(
- setFailUnfulfillableRequestConsumer, getRequiredResourcesSupplier);
+ setFailUnfulfillableRequestConsumer,
+ getRequiredResourcesSupplier,
+ processRequirementsConsumer,
+ clearRequirementsConsumer);
}
}