You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tr...@apache.org on 2019/06/14 12:36:49 UTC
[flink] 01/02: [FLINK-11059][runtime] Add slot reconciliation
between JM and TE
This is an automated email from the ASF dual-hosted git repository.
trohrmann pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit a4865cd67239c938e0075bcc2b112b944cc3c249
Author: shuai-xu <sh...@foxmail.com>
AuthorDate: Tue Dec 4 12:45:34 2018 +0800
[FLINK-11059][runtime] Add slot reconciliation between JM and TE
With this commit we introduce an AllocatedSlotReport which is periodically sent from the
JM to the TE. This report is then used to reconcile the JM's and the TE's view on the state
of allocated slots.
Furthermore, with this commit we drop slots which could not be freed via
TaskExecutorGateway#freeSlot and instead rely on the heartbeat reconciliation logic.
This closes #7227.
---
.../flink/runtime/jobmaster/AllocatedSlotInfo.java | 60 ++++++++++
.../runtime/jobmaster/AllocatedSlotReport.java | 62 ++++++++++
.../apache/flink/runtime/jobmaster/JobMaster.java | 16 +--
.../flink/runtime/jobmaster/slotpool/SlotPool.java | 9 ++
.../runtime/jobmaster/slotpool/SlotPoolImpl.java | 34 ++++--
.../flink/runtime/taskexecutor/TaskExecutor.java | 77 +++++++++++-
.../runtime/taskexecutor/TaskExecutorGateway.java | 3 +-
.../flink/runtime/jobmaster/JobMasterTest.java | 9 +-
.../runtime/jobmaster/slotpool/SlotPoolTest.java | 131 +++++++++++++++------
.../runtime/taskexecutor/TaskExecutorTest.java | 82 +++++++++++++
.../taskexecutor/TestingTaskExecutorGateway.java | 9 +-
.../TestingTaskExecutorGatewayBuilder.java | 7 +-
12 files changed, 425 insertions(+), 74 deletions(-)
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/AllocatedSlotInfo.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/AllocatedSlotInfo.java
new file mode 100644
index 0000000..3e587df
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/AllocatedSlotInfo.java
@@ -0,0 +1,60 @@
+/*
+ * 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.jobmaster;
+
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+
+import java.io.Serializable;
+
+import static org.apache.flink.util.Preconditions.checkArgument;
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * Information about an allocated slot which is owned by a JobMaster.
+ */
+public class AllocatedSlotInfo implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final int slotIndex;
+
+ private final AllocationID allocationId;
+
+ public AllocatedSlotInfo(int index, AllocationID allocationId) {
+ checkArgument(index >= 0);
+ this.slotIndex = index;
+ this.allocationId = checkNotNull(allocationId);
+ }
+
+ public AllocationID getAllocationId() {
+ return allocationId;
+ }
+
+ public int getSlotIndex() {
+ return slotIndex;
+ }
+
+ @Override
+ public String toString() {
+ return "AllocatedSlotInfo{" +
+ "slotIndex=" + slotIndex +
+ ", allocationId=" + allocationId +
+ '}';
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/AllocatedSlotReport.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/AllocatedSlotReport.java
new file mode 100644
index 0000000..ef1d42b
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/AllocatedSlotReport.java
@@ -0,0 +1,62 @@
+/*
+ * 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.jobmaster;
+
+import org.apache.flink.api.common.JobID;
+
+import java.io.Serializable;
+import java.util.Collection;
+import java.util.Collections;
+
+import static org.apache.flink.util.Preconditions.checkNotNull;
+
+/**
+ * The report of currently allocated slots from a given TaskExecutor by a JobMaster.
+ * This report is sent periodically to the TaskExecutor in order to reconcile the internal state of slot allocations.
+ */
+public class AllocatedSlotReport implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private final JobID jobId;
+
+ /** The allocated slots in slot pool. */
+ private final Collection<AllocatedSlotInfo> allocatedSlotInfos;
+
+ public AllocatedSlotReport(JobID jobId, Collection<AllocatedSlotInfo> allocatedSlotInfos) {
+ this.jobId = checkNotNull(jobId);
+ this.allocatedSlotInfos = checkNotNull(allocatedSlotInfos);
+ }
+
+ public JobID getJobId() {
+ return jobId;
+ }
+
+ public Collection<AllocatedSlotInfo> getAllocatedSlotInfos() {
+ return Collections.unmodifiableCollection(allocatedSlotInfos);
+ }
+
+ @Override
+ public String toString() {
+ return "AllocatedSlotReport{" +
+ "jobId=" + jobId +
+ ", allocatedSlotInfos=" + allocatedSlotInfos +
+ '}';
+ }
+}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
index cc7139b..1b1b991 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMaster.java
@@ -140,7 +140,7 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
private final JobManagerJobMetricGroupFactory jobMetricGroupFactory;
- private final HeartbeatManager<AccumulatorReport, Void> taskManagerHeartbeatManager;
+ private final HeartbeatManager<AccumulatorReport, AllocatedSlotReport> taskManagerHeartbeatManager;
private final HeartbeatManager<Void, Void> resourceManagerHeartbeatManager;
@@ -559,15 +559,15 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
registeredTaskManagers.put(taskManagerId, Tuple2.of(taskManagerLocation, taskExecutorGateway));
// monitor the task manager as heartbeat target
- taskManagerHeartbeatManager.monitorTarget(taskManagerId, new HeartbeatTarget<Void>() {
+ taskManagerHeartbeatManager.monitorTarget(taskManagerId, new HeartbeatTarget<AllocatedSlotReport>() {
@Override
- public void receiveHeartbeat(ResourceID resourceID, Void payload) {
+ public void receiveHeartbeat(ResourceID resourceID, AllocatedSlotReport payload) {
// the task manager will not request heartbeat, so this method will never be called currently
}
@Override
- public void requestHeartbeat(ResourceID resourceID, Void payload) {
- taskExecutorGateway.heartbeatFromJobManager(resourceID);
+ public void requestHeartbeat(ResourceID resourceID, AllocatedSlotReport allocatedSlotReport) {
+ taskExecutorGateway.heartbeatFromJobManager(resourceID, allocatedSlotReport);
}
});
@@ -1096,7 +1096,7 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
}
}
- private class TaskManagerHeartbeatListener implements HeartbeatListener<AccumulatorReport, Void> {
+ private class TaskManagerHeartbeatListener implements HeartbeatListener<AccumulatorReport, AllocatedSlotReport> {
private final JobMasterGateway jobMasterGateway;
@@ -1119,8 +1119,8 @@ public class JobMaster extends FencedRpcEndpoint<JobMasterId> implements JobMast
}
@Override
- public CompletableFuture<Void> retrievePayload(ResourceID resourceID) {
- return CompletableFuture.completedFuture(null);
+ public CompletableFuture<AllocatedSlotReport> retrievePayload(ResourceID resourceID) {
+ return callAsync(() -> slotPool.createAllocatedSlotReport(resourceID), RpcUtils.INF_TIMEOUT);
}
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java
index 2191cea..c33e71d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPool.java
@@ -24,6 +24,7 @@ import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
import org.apache.flink.runtime.concurrent.ComponentMainThreadExecutor;
import org.apache.flink.runtime.jobmanager.slots.TaskManagerGateway;
+import org.apache.flink.runtime.jobmaster.AllocatedSlotReport;
import org.apache.flink.runtime.jobmaster.JobMasterId;
import org.apache.flink.runtime.jobmaster.SlotInfo;
import org.apache.flink.runtime.jobmaster.SlotRequestId;
@@ -163,4 +164,12 @@ public interface SlotPool extends AllocatedSlotActions, AutoCloseable {
@Nonnull SlotRequestId slotRequestId,
@Nonnull ResourceProfile resourceProfile,
@RpcTimeout Time timeout);
+
+ /**
+ * Create report about the allocated slots belonging to the specified task manager.
+ *
+ * @param taskManagerId identifies the task manager
+ * @return the allocated slots on the task manager
+ */
+ AllocatedSlotReport createAllocatedSlotReport(ResourceID taskManagerId);
}
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolImpl.java b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolImpl.java
index 0671e4e..d31976e 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolImpl.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolImpl.java
@@ -32,9 +32,11 @@ import org.apache.flink.runtime.concurrent.FutureUtils;
import org.apache.flink.runtime.executiongraph.ExecutionGraph;
import org.apache.flink.runtime.jobmanager.scheduler.NoResourceAvailableException;
import org.apache.flink.runtime.jobmanager.slots.TaskManagerGateway;
+import org.apache.flink.runtime.jobmaster.AllocatedSlotInfo;
import org.apache.flink.runtime.jobmaster.JobMasterId;
import org.apache.flink.runtime.jobmaster.SlotInfo;
import org.apache.flink.runtime.jobmaster.SlotRequestId;
+import org.apache.flink.runtime.jobmaster.AllocatedSlotReport;
import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
import org.apache.flink.runtime.resourcemanager.SlotRequest;
@@ -45,6 +47,8 @@ import org.apache.flink.runtime.util.clock.SystemClock;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.Preconditions;
+import org.apache.flink.shaded.guava18.com.google.common.collect.Iterables;
+
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -715,6 +719,20 @@ public class SlotPoolImpl implements SlotPool {
}
}
+ @Override
+ public AllocatedSlotReport createAllocatedSlotReport(ResourceID taskManagerId) {
+ final Set<AllocatedSlot> availableSlotsForTaskManager = availableSlots.getSlotsForTaskManager(taskManagerId);
+ final Set<AllocatedSlot> allocatedSlotsForTaskManager = allocatedSlots.getSlotsForTaskManager(taskManagerId);
+
+ List<AllocatedSlotInfo> allocatedSlotInfos = new ArrayList<>(
+ availableSlotsForTaskManager.size() + allocatedSlotsForTaskManager.size());
+ for (AllocatedSlot allocatedSlot : Iterables.concat(availableSlotsForTaskManager, allocatedSlotsForTaskManager)) {
+ allocatedSlotInfos.add(
+ new AllocatedSlotInfo(allocatedSlot.getPhysicalSlotNumber(), allocatedSlot.getAllocationId()));
+ }
+ return new AllocatedSlotReport(jobId, allocatedSlotInfos);
+ }
+
// ------------------------------------------------------------------------
// Internal methods
// ------------------------------------------------------------------------
@@ -773,15 +791,9 @@ public class SlotPoolImpl implements SlotPool {
componentMainThreadExecutor,
(Acknowledge ignored, Throwable throwable) -> {
if (throwable != null) {
- if (registeredTaskManagers.contains(expiredSlot.getTaskManagerId())) {
- log.debug("Releasing slot [{}] of registered TaskExecutor {} failed. " +
- "Trying to fulfill a different slot request.", allocationID, expiredSlot.getTaskManagerId(),
- throwable);
- tryFulfillSlotRequestOrMakeAvailable(expiredSlot);
- } else {
- log.debug("Releasing slot [{}] failed and owning TaskExecutor {} is no " +
- "longer registered. Discarding slot.", allocationID, expiredSlot.getTaskManagerId());
- }
+ // The slot status will be synced to task manager in next heartbeat.
+ log.debug("Releasing slot [{}] of registered TaskExecutor {} failed. Discarding slot.",
+ allocationID, expiredSlot.getTaskManagerId(), throwable);
}
});
}
@@ -1199,6 +1211,10 @@ public class SlotPoolImpl implements SlotPool {
availableSlotsByTaskManager.clear();
availableSlotsByHost.clear();
}
+
+ Set<AllocatedSlot> getSlotsForTaskManager(ResourceID resourceId) {
+ return availableSlotsByTaskManager.getOrDefault(resourceId, Collections.emptySet());
+ }
}
// ------------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
index 6899b32..4238d71 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
@@ -48,10 +48,11 @@ import org.apache.flink.runtime.heartbeat.HeartbeatTarget;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
import org.apache.flink.runtime.instance.HardwareDescription;
import org.apache.flink.runtime.instance.InstanceID;
-import org.apache.flink.runtime.shuffle.ShuffleEnvironment;
import org.apache.flink.runtime.io.network.partition.ResultPartitionConsumableNotifier;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.jobgraph.tasks.InputSplitProvider;
+import org.apache.flink.runtime.jobmaster.AllocatedSlotInfo;
+import org.apache.flink.runtime.jobmaster.AllocatedSlotReport;
import org.apache.flink.runtime.jobmaster.JMTMRegistrationSuccess;
import org.apache.flink.runtime.jobmaster.JobMasterGateway;
import org.apache.flink.runtime.jobmaster.JobMasterId;
@@ -72,6 +73,7 @@ import org.apache.flink.runtime.rpc.FatalErrorHandler;
import org.apache.flink.runtime.rpc.RpcEndpoint;
import org.apache.flink.runtime.rpc.RpcService;
import org.apache.flink.runtime.rpc.akka.AkkaRpcServiceUtils;
+import org.apache.flink.runtime.shuffle.ShuffleEnvironment;
import org.apache.flink.runtime.state.TaskExecutorLocalStateStoresManager;
import org.apache.flink.runtime.state.TaskLocalStateStore;
import org.apache.flink.runtime.state.TaskStateManager;
@@ -104,6 +106,8 @@ import org.apache.flink.types.SerializableOptional;
import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkException;
+import org.apache.flink.shaded.guava18.com.google.common.collect.Sets;
+
import javax.annotation.Nonnull;
import javax.annotation.Nullable;
@@ -120,9 +124,11 @@ import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
import static org.apache.flink.util.Preconditions.checkArgument;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -144,7 +150,7 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
private final TaskManagerConfiguration taskManagerConfiguration;
/** The heartbeat manager for job manager in the task manager. */
- private final HeartbeatManager<Void, AccumulatorReport> jobManagerHeartbeatManager;
+ private final HeartbeatManager<AllocatedSlotReport, AccumulatorReport> jobManagerHeartbeatManager;
/** The heartbeat manager for resource manager in the task manager. */
private final HeartbeatManager<Void, SlotReport> resourceManagerHeartbeatManager;
@@ -656,7 +662,8 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
// ----------------------------------------------------------------------
@Override
- public void heartbeatFromJobManager(ResourceID resourceID) {
+ public void heartbeatFromJobManager(ResourceID resourceID, AllocatedSlotReport allocatedSlotReport) {
+ jobManagerHeartbeatManager.receiveHeartbeat(resourceID, allocatedSlotReport);
jobManagerHeartbeatManager.requestHeartbeat(resourceID, null);
}
@@ -1410,6 +1417,64 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
}
}
+ /**
+ * Syncs the TaskExecutor's view on its allocated slots with the JobMaster's view.
+ * Slots which are no longer reported by the JobMaster are being freed.
+ * Slots which the JobMaster thinks it still owns but which are no longer allocated to it
+ * will be failed via {@link JobMasterGateway#failSlot}.
+ *
+ * @param allocatedSlotReport represents the JobMaster's view on the current slot allocation state
+ */
+ private void syncSlotsWithSnapshotFromJobMaster(AllocatedSlotReport allocatedSlotReport) {
+ JobManagerConnection jobManagerConnection = jobManagerTable.get(allocatedSlotReport.getJobId());
+ if (jobManagerConnection != null) {
+ final JobMasterGateway jobMasterGateway = jobManagerConnection.getJobManagerGateway();
+
+ failNoLongerAllocatedSlots(allocatedSlotReport, jobMasterGateway);
+
+ freeNoLongerUsedSlots(allocatedSlotReport);
+ } else {
+ log.debug("Ignoring allocated slot report from job {} because there is no active leader.",
+ allocatedSlotReport.getJobId());
+ }
+ }
+
+ private void failNoLongerAllocatedSlots(AllocatedSlotReport allocatedSlotReport, JobMasterGateway jobMasterGateway) {
+ for (AllocatedSlotInfo allocatedSlotInfo : allocatedSlotReport.getAllocatedSlotInfos()) {
+ final AllocationID allocationId = allocatedSlotInfo.getAllocationId();
+ if (!taskSlotTable.isAllocated(
+ allocatedSlotInfo.getSlotIndex(),
+ allocatedSlotReport.getJobId(),
+ allocationId)) {
+ jobMasterGateway.failSlot(
+ getResourceID(),
+ allocationId,
+ new FlinkException(
+ String.format(
+ "Slot %s on TaskExecutor %s is not allocated by job %s.",
+ allocatedSlotInfo.getSlotIndex(),
+ getResourceID(),
+ allocatedSlotReport.getJobId())));
+ }
+ }
+ }
+
+ private void freeNoLongerUsedSlots(AllocatedSlotReport allocatedSlotReport) {
+ final Iterator<AllocationID> slotsTaskManagerSide = taskSlotTable.getActiveSlots(allocatedSlotReport.getJobId());
+ final Set<AllocationID> activeSlots = Sets.newHashSet(slotsTaskManagerSide);
+ final Set<AllocationID> reportedSlots = allocatedSlotReport.getAllocatedSlotInfos().stream()
+ .map(AllocatedSlotInfo::getAllocationId).collect(Collectors.toSet());
+
+ final Sets.SetView<AllocationID> difference = Sets.difference(activeSlots, reportedSlots);
+
+ for (AllocationID allocationID : difference) {
+ freeSlotInternal(
+ allocationID,
+ new FlinkException(
+ String.format("%s is no longer allocated by job %s.", allocationID, allocatedSlotReport.getJobId())));
+ }
+ }
+
// ------------------------------------------------------------------------
// Internal utility methods
// ------------------------------------------------------------------------
@@ -1599,7 +1664,7 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
}
}
- private class JobManagerHeartbeatListener implements HeartbeatListener<Void, AccumulatorReport> {
+ private class JobManagerHeartbeatListener implements HeartbeatListener<AllocatedSlotReport, AccumulatorReport> {
@Override
public void notifyHeartbeatTimeout(final ResourceID resourceID) {
@@ -1621,8 +1686,8 @@ public class TaskExecutor extends RpcEndpoint implements TaskExecutorGateway {
}
@Override
- public void reportPayload(ResourceID resourceID, Void payload) {
- // nothing to do since the payload is of type Void
+ public void reportPayload(ResourceID resourceID, AllocatedSlotReport allocatedSlotReport) {
+ runAsync(() -> syncSlotsWithSnapshotFromJobMaster(allocatedSlotReport));
}
@Override
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGateway.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGateway.java
index 5ab16dc..75290d3 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGateway.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGateway.java
@@ -30,6 +30,7 @@ import org.apache.flink.runtime.deployment.TaskDeploymentDescriptor;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.executiongraph.PartitionInfo;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.jobmaster.AllocatedSlotReport;
import org.apache.flink.runtime.jobmaster.JobMasterId;
import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.runtime.messages.StackTraceSampleResponse;
@@ -152,7 +153,7 @@ public interface TaskExecutorGateway extends RpcGateway {
*
* @param heartbeatOrigin unique id of the job manager
*/
- void heartbeatFromJobManager(ResourceID heartbeatOrigin);
+ void heartbeatFromJobManager(ResourceID heartbeatOrigin, AllocatedSlotReport allocatedSlotReport);
/**
* Heartbeat request from the resource manager.
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
index f8a05fd..af447c7 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/JobMasterTest.java
@@ -160,7 +160,6 @@ import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiConsumer;
import java.util.function.BiFunction;
-import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
@@ -338,7 +337,7 @@ public class JobMasterTest extends TestLogger {
final CompletableFuture<JobID> disconnectedJobManagerFuture = new CompletableFuture<>();
final TaskManagerLocation taskManagerLocation = new LocalTaskManagerLocation();
final TestingTaskExecutorGateway taskExecutorGateway = new TestingTaskExecutorGatewayBuilder()
- .setHeartbeatJobManagerConsumer(heartbeatResourceIdFuture::complete)
+ .setHeartbeatJobManagerConsumer((taskManagerId, ignored) -> heartbeatResourceIdFuture.complete(taskManagerId))
.setDisconnectJobManagerConsumer((jobId, throwable) -> disconnectedJobManagerFuture.complete(jobId))
.createTestingTaskExecutorGateway();
@@ -1579,7 +1578,7 @@ public class JobMasterTest extends TestLogger {
(localTaskManagerLocation, jobMasterGateway) -> jobMasterGateway.disconnectTaskManager(
localTaskManagerLocation.getResourceID(),
new FlinkException("Test disconnectTaskManager exception.")),
- (jobMasterGateway, resourceID) -> ignored -> {});
+ (jobMasterGateway, resourceID) -> (ignoredA, ignoredB) -> {});
}
@Test
@@ -1588,7 +1587,7 @@ public class JobMasterTest extends TestLogger {
runJobFailureWhenTaskExecutorTerminatesTest(
fastHeartbeatServices,
(localTaskManagerLocation, jobMasterGateway) -> respondToHeartbeats.set(false),
- (jobMasterGateway, taskManagerResourceId) -> resourceId -> {
+ (jobMasterGateway, taskManagerResourceId) -> (resourceId, ignored) -> {
if (respondToHeartbeats.get()) {
jobMasterGateway.heartbeatFromTaskManager(taskManagerResourceId, new AccumulatorReport(Collections.emptyList()));
}
@@ -1599,7 +1598,7 @@ public class JobMasterTest extends TestLogger {
private void runJobFailureWhenTaskExecutorTerminatesTest(
HeartbeatServices heartbeatServices,
BiConsumer<LocalTaskManagerLocation, JobMasterGateway> jobReachedRunningState,
- BiFunction<JobMasterGateway, ResourceID, Consumer<ResourceID>> heartbeatConsumerFunction) throws Exception {
+ BiFunction<JobMasterGateway, ResourceID, BiConsumer<ResourceID, AllocatedSlotReport>> heartbeatConsumerFunction) throws Exception {
final JobGraph jobGraph = createSingleVertexJobGraph();
final TestingOnCompletionActions onCompletionActions = new TestingOnCompletionActions();
final JobMaster jobMaster = createJobMaster(
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolTest.java
index d66027f..cc95b04 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/jobmaster/slotpool/SlotPoolTest.java
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.jobmaster.slotpool;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.time.Time;
+import org.apache.flink.core.testutils.OneShotLatch;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
@@ -31,6 +32,8 @@ import org.apache.flink.runtime.executiongraph.utils.SimpleAckingTaskManagerGate
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobmanager.scheduler.DummyScheduledUnit;
import org.apache.flink.runtime.jobmanager.scheduler.ScheduledUnit;
+import org.apache.flink.runtime.jobmaster.AllocatedSlotInfo;
+import org.apache.flink.runtime.jobmaster.AllocatedSlotReport;
import org.apache.flink.runtime.jobmaster.JobMasterId;
import org.apache.flink.runtime.jobmaster.LogicalSlot;
import org.apache.flink.runtime.jobmaster.SlotRequestId;
@@ -48,11 +51,13 @@ import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.FlinkException;
import org.apache.flink.util.TestLogger;
+import org.hamcrest.Description;
+import org.hamcrest.Matcher;
import org.hamcrest.Matchers;
+import org.hamcrest.TypeSafeDiagnosingMatcher;
import org.junit.Before;
import org.junit.Test;
-import java.util.ArrayDeque;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -66,9 +71,11 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.stream.Collectors;
import static org.apache.flink.runtime.jobmaster.slotpool.AvailableSlotsTest.DEFAULT_TESTING_PROFILE;
import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.containsInAnyOrder;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.is;
import static org.junit.Assert.assertEquals;
@@ -599,11 +606,10 @@ public class SlotPoolTest extends TestLogger {
}
/**
- * Tests that idle slots which cannot be released are only recycled if the owning {@link TaskExecutor}
- * is still registered at the {@link SlotPoolImpl}. See FLINK-9047.
+ * Tests that idle slots which cannot be released will be discarded. See FLINK-11059.
*/
@Test
- public void testReleasingIdleSlotFailed() throws Exception {
+ public void testDiscardIdleSlotIfReleasingFailed() throws Exception {
final ManualClock clock = new ManualClock();
try (SlotPoolImpl slotPool = new SlotPoolImpl(
@@ -618,20 +624,12 @@ public class SlotPoolTest extends TestLogger {
final AllocationID expiredAllocationId = new AllocationID();
final SlotOffer slotToExpire = new SlotOffer(expiredAllocationId, 0, ResourceProfile.UNKNOWN);
- final ArrayDeque<CompletableFuture<Acknowledge>> responseQueue = new ArrayDeque<>(2);
+ OneShotLatch freeSlotLatch = new OneShotLatch();
taskManagerGateway.setFreeSlotFunction((AllocationID allocationId, Throwable cause) -> {
- if (responseQueue.isEmpty()) {
- return CompletableFuture.completedFuture(Acknowledge.get());
- } else {
- return responseQueue.pop();
- }
+ freeSlotLatch.trigger();
+ return FutureUtils.completedExceptionally(new TimeoutException("Test failure"));
});
- responseQueue.add(FutureUtils.completedExceptionally(new FlinkException("Test failure")));
-
- final CompletableFuture<Acknowledge> responseFuture = new CompletableFuture<>();
- responseQueue.add(responseFuture);
-
assertThat(slotPool.registerTaskManager(taskManagerLocation.getResourceID()), Matchers.is(true));
assertThat(slotPool.offerSlot(taskManagerLocation, taskManagerGateway, slotToExpire), Matchers.is(true));
@@ -640,29 +638,9 @@ public class SlotPoolTest extends TestLogger {
slotPool.triggerCheckIdleSlot();
- CompletableFuture<LogicalSlot> allocatedSlotFuture = allocateSlot(scheduler, new SlotRequestId());
-
- // wait until the slot has been fulfilled with the previously idling slot
- final LogicalSlot logicalSlot = allocatedSlotFuture.get();
- assertThat(logicalSlot.getAllocationId(), Matchers.is(expiredAllocationId));
-
- // return the slot
- scheduler.returnLogicalSlot(logicalSlot);
+ freeSlotLatch.await();
- // advance the time so that the returned slot is now idling
- clock.advanceTime(timeout.toMilliseconds() + 1, TimeUnit.MILLISECONDS);
-
- slotPool.triggerCheckIdleSlot();
-
- // request a new slot after the idling slot has been released
- allocatedSlotFuture = allocateSlot(scheduler, new SlotRequestId());
-
- // release the TaskExecutor before we get a response from the slot releasing
- slotPool.releaseTaskManager(taskManagerLocation.getResourceID(), null);
-
- // let the slot releasing fail --> since the owning TaskExecutor is no longer registered
- // the slot should be discarded
- responseFuture.completeExceptionally(new FlinkException("Second test exception"));
+ CompletableFuture<LogicalSlot> allocatedSlotFuture = allocateSlot(scheduler, new SlotRequestId());
try {
// since the slot must have been discarded, we cannot fulfill the slot request
@@ -670,8 +648,8 @@ public class SlotPoolTest extends TestLogger {
fail("Expected to fail with a timeout.");
} catch (TimeoutException ignored) {
// expected
+ assertEquals(0, slotPool.getAvailableSlots().size());
}
-
}
}
@@ -770,6 +748,83 @@ public class SlotPoolTest extends TestLogger {
}
}
+ /**
+ * Tests that create report of allocated slots on a {@link TaskExecutor}.
+ */
+ @Test
+ public void testCreateAllocatedSlotReport() throws Exception {
+
+ try (SlotPoolImpl slotPool = new SlotPoolImpl(jobId)) {
+
+ final ArrayBlockingQueue<AllocationID> allocationIds = new ArrayBlockingQueue<>(1);
+ resourceManagerGateway.setRequestSlotConsumer(
+ slotRequest -> allocationIds.offer(slotRequest.getAllocationId()));
+
+ setupSlotPool(slotPool, resourceManagerGateway, mainThreadExecutor);
+ Scheduler scheduler = setupScheduler(slotPool, mainThreadExecutor);
+
+ final SlotRequestId slotRequestId = new SlotRequestId();
+ final CompletableFuture<LogicalSlot> slotRequestFuture = allocateSlot(scheduler, slotRequestId);
+
+ final List<AllocatedSlotInfo> allocatedSlotInfos = new ArrayList<>(2);
+ final List<SlotOffer> slotOffers = new ArrayList<>(2);
+
+ final AllocationID allocatedId = allocationIds.take();
+ slotOffers.add(new SlotOffer(allocatedId, 0, ResourceProfile.UNKNOWN));
+ allocatedSlotInfos.add(new AllocatedSlotInfo(0, allocatedId));
+
+ final AllocationID availableId = new AllocationID();
+ slotOffers.add(new SlotOffer(availableId, 1, ResourceProfile.UNKNOWN));
+ allocatedSlotInfos.add(new AllocatedSlotInfo(1, availableId));
+
+ slotPool.registerTaskManager(taskManagerLocation.getResourceID());
+ slotPool.offerSlots(taskManagerLocation, taskManagerGateway, slotOffers);
+
+ // wait for the completion of slot future
+ slotRequestFuture.get();
+
+ final AllocatedSlotReport slotReport = slotPool.createAllocatedSlotReport(taskManagerLocation.getResourceID());
+ assertThat(jobId, is(slotReport.getJobId()));
+ assertThat(slotReport.getAllocatedSlotInfos(), containsInAnyOrder(isEachEqual(allocatedSlotInfos)));
+ }
+ }
+
+ private static Collection<Matcher<? super AllocatedSlotInfo>> isEachEqual(Collection<AllocatedSlotInfo> allocatedSlotInfos) {
+ return allocatedSlotInfos
+ .stream()
+ .map(SlotPoolTest::isEqualAllocatedSlotInfo)
+ .collect(Collectors.toList());
+ }
+
+ private static Matcher<AllocatedSlotInfo> isEqualAllocatedSlotInfo(AllocatedSlotInfo expectedAllocatedSlotInfo) {
+ return new TypeSafeDiagnosingMatcher<AllocatedSlotInfo>() {
+ @Override
+ public void describeTo(Description description) {
+ description.appendText(describeAllocatedSlotInformation(expectedAllocatedSlotInfo));
+ }
+
+ private String describeAllocatedSlotInformation(AllocatedSlotInfo expectedAllocatedSlotInformation) {
+ return expectedAllocatedSlotInformation.toString();
+ }
+
+ @Override
+ protected boolean matchesSafely(AllocatedSlotInfo item, Description mismatchDescription) {
+ final boolean matches = item.getAllocationId().equals(expectedAllocatedSlotInfo.getAllocationId()) &&
+ item.getSlotIndex() == expectedAllocatedSlotInfo.getSlotIndex();
+
+ if (!matches) {
+ mismatchDescription
+ .appendText("Actual value ")
+ .appendText(describeAllocatedSlotInformation(item))
+ .appendText(" differs from expected value ")
+ .appendText(describeAllocatedSlotInformation(expectedAllocatedSlotInfo));
+ }
+
+ return matches;
+ }
+ };
+ }
+
private CompletableFuture<LogicalSlot> allocateSlot(Scheduler scheduler, SlotRequestId slotRequestId) {
return scheduler.allocateSlot(
slotRequestId,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java
index ff369e6..31b0f24 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TaskExecutorTest.java
@@ -62,6 +62,8 @@ import org.apache.flink.runtime.io.network.NettyShuffleEnvironmentBuilder;
import org.apache.flink.runtime.io.network.partition.NoOpResultPartitionConsumableNotifier;
import org.apache.flink.runtime.jobgraph.JobVertexID;
import org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable;
+import org.apache.flink.runtime.jobmaster.AllocatedSlotInfo;
+import org.apache.flink.runtime.jobmaster.AllocatedSlotReport;
import org.apache.flink.runtime.jobmaster.JMTMRegistrationSuccess;
import org.apache.flink.runtime.jobmaster.JobMasterGateway;
import org.apache.flink.runtime.jobmaster.JobMasterId;
@@ -124,6 +126,7 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
+import java.util.List;
import java.util.Queue;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
@@ -144,6 +147,7 @@ import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.instanceOf;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;
+import static org.hamcrest.Matchers.nullValue;
import static org.hamcrest.Matchers.startsWith;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -1662,6 +1666,84 @@ public class TaskExecutorTest extends TestLogger {
}
}
+ /**
+ * Tests that the TaskExecutor syncs its slots view with the JobMaster's view
+ * via the AllocatedSlotReport reported by the heartbeat (See FLINK-11059).
+ */
+ @Test(timeout = 10000L)
+ public void testSyncSlotsWithJobMasterByHeartbeat() throws Exception {
+ final TaskSlotTable taskSlotTable = new TaskSlotTable(
+ Arrays.asList(ResourceProfile.UNKNOWN, ResourceProfile.UNKNOWN),
+ timerService);
+ final TaskManagerServices taskManagerServices = new TaskManagerServicesBuilder().setTaskSlotTable(taskSlotTable).build();
+
+ final TaskExecutor taskExecutor = createTaskExecutor(taskManagerServices);
+
+ final TestingResourceManagerGateway testingResourceManagerGateway = new TestingResourceManagerGateway();
+
+ final BlockingQueue<AllocationID> allocationsNotifiedFree = new ArrayBlockingQueue<>(2);
+
+ OneShotLatch initialSlotReporting = new OneShotLatch();
+ testingResourceManagerGateway.setSendSlotReportFunction(resourceIDInstanceIDSlotReportTuple3 -> {
+ initialSlotReporting.trigger();
+ return CompletableFuture.completedFuture(Acknowledge.get());
+
+ });
+
+ testingResourceManagerGateway.setNotifySlotAvailableConsumer(instanceIDSlotIDAllocationIDTuple3 ->
+ allocationsNotifiedFree.offer(instanceIDSlotIDAllocationIDTuple3.f2));
+
+ rpc.registerGateway(testingResourceManagerGateway.getAddress(), testingResourceManagerGateway);
+ resourceManagerLeaderRetriever.notifyListener(testingResourceManagerGateway.getAddress(), testingResourceManagerGateway.getFencingToken().toUUID());
+
+ final CountDownLatch slotOfferings = new CountDownLatch(2);
+ final BlockingQueue<AllocationID> failedSlotFutures = new ArrayBlockingQueue<>(2);
+ final TestingJobMasterGateway jobMasterGateway = new TestingJobMasterGatewayBuilder()
+ .setFailSlotConsumer((resourceID, allocationID, throwable) ->
+ failedSlotFutures.offer(allocationID))
+ .setOfferSlotsFunction((resourceID, slotOffers) -> {
+ slotOffers.forEach((ignored) -> slotOfferings.countDown());
+ return CompletableFuture.completedFuture(new ArrayList<>(slotOffers));
+ })
+ .build();
+ final String jobManagerAddress = jobMasterGateway.getAddress();
+ rpc.registerGateway(jobManagerAddress, jobMasterGateway);
+ jobManagerLeaderRetriever.notifyListener(jobManagerAddress, jobMasterGateway.getFencingToken().toUUID());
+
+ taskExecutor.start();
+
+ try {
+ final TaskExecutorGateway taskExecutorGateway = taskExecutor.getSelfGateway(TaskExecutorGateway.class);
+
+ initialSlotReporting.await();
+
+ final SlotID slotId1 = new SlotID(taskExecutor.getResourceID(), 0);
+ final SlotID slotId2 = new SlotID(taskExecutor.getResourceID(), 1);
+ final AllocationID allocationIdInBoth = new AllocationID();
+ final AllocationID allocationIdOnlyInJM = new AllocationID();
+ final AllocationID allocationIdOnlyInTM = new AllocationID();
+
+ taskExecutorGateway.requestSlot(slotId1, jobId, allocationIdInBoth, "foobar", testingResourceManagerGateway.getFencingToken(), timeout);
+ taskExecutorGateway.requestSlot(slotId2, jobId, allocationIdOnlyInTM, "foobar", testingResourceManagerGateway.getFencingToken(), timeout);
+
+ slotOfferings.await();
+
+ List<AllocatedSlotInfo> allocatedSlotInfos = Arrays.asList(
+ new AllocatedSlotInfo(0, allocationIdInBoth),
+ new AllocatedSlotInfo(1, allocationIdOnlyInJM)
+ );
+ AllocatedSlotReport allocatedSlotReport = new AllocatedSlotReport(jobId, allocatedSlotInfos);
+ taskExecutorGateway.heartbeatFromJobManager(ResourceID.generate(), allocatedSlotReport);
+
+ assertThat(failedSlotFutures.take(), is(allocationIdOnlyInJM));
+ assertThat(allocationsNotifiedFree.take(), is(allocationIdOnlyInTM));
+ assertThat(failedSlotFutures.poll(5L, TimeUnit.MILLISECONDS), nullValue());
+ assertThat(allocationsNotifiedFree.poll(5L, TimeUnit.MILLISECONDS), nullValue());
+ } finally {
+ RpcUtils.terminateRpcEndpoint(taskExecutor, timeout);
+ }
+ }
+
private TaskExecutorLocalStateStoresManager createTaskExecutorLocalStateStoresManager() throws IOException {
return new TaskExecutorLocalStateStoresManager(
false,
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGateway.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGateway.java
index 1021a77..844ef0d 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGateway.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGateway.java
@@ -31,6 +31,7 @@ import org.apache.flink.runtime.deployment.TaskDeploymentDescriptor;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
import org.apache.flink.runtime.executiongraph.PartitionInfo;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
+import org.apache.flink.runtime.jobmaster.AllocatedSlotReport;
import org.apache.flink.runtime.jobmaster.JobMasterId;
import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.runtime.messages.StackTraceSampleResponse;
@@ -55,7 +56,7 @@ public class TestingTaskExecutorGateway implements TaskExecutorGateway {
private final String hostname;
- private final Consumer<ResourceID> heartbeatJobManagerConsumer;
+ private final BiConsumer<ResourceID, AllocatedSlotReport> heartbeatJobManagerConsumer;
private final BiConsumer<JobID, Throwable> disconnectJobManagerConsumer;
@@ -76,7 +77,7 @@ public class TestingTaskExecutorGateway implements TaskExecutorGateway {
TestingTaskExecutorGateway(
String address,
String hostname,
- Consumer<ResourceID> heartbeatJobManagerConsumer,
+ BiConsumer<ResourceID, AllocatedSlotReport> heartbeatJobManagerConsumer,
BiConsumer<JobID, Throwable> disconnectJobManagerConsumer,
BiFunction<TaskDeploymentDescriptor, JobMasterId, CompletableFuture<Acknowledge>> submitTaskConsumer,
Function<Tuple5<SlotID, JobID, AllocationID, String, ResourceManagerId>, CompletableFuture<Acknowledge>> requestSlotFunction,
@@ -145,8 +146,8 @@ public class TestingTaskExecutorGateway implements TaskExecutorGateway {
}
@Override
- public void heartbeatFromJobManager(ResourceID heartbeatOrigin) {
- heartbeatJobManagerConsumer.accept(heartbeatOrigin);
+ public void heartbeatFromJobManager(ResourceID heartbeatOrigin, AllocatedSlotReport allocatedSlotReport) {
+ heartbeatJobManagerConsumer.accept(heartbeatOrigin, allocatedSlotReport);
}
@Override
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGatewayBuilder.java b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGatewayBuilder.java
index 7bc73cf..483f285 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGatewayBuilder.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/taskexecutor/TestingTaskExecutorGatewayBuilder.java
@@ -25,6 +25,7 @@ import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.clusterframework.types.SlotID;
import org.apache.flink.runtime.deployment.TaskDeploymentDescriptor;
import org.apache.flink.runtime.executiongraph.ExecutionAttemptID;
+import org.apache.flink.runtime.jobmaster.AllocatedSlotReport;
import org.apache.flink.runtime.jobmaster.JobMasterId;
import org.apache.flink.runtime.messages.Acknowledge;
import org.apache.flink.runtime.resourcemanager.ResourceManagerId;
@@ -41,7 +42,7 @@ import java.util.function.Supplier;
*/
public class TestingTaskExecutorGatewayBuilder {
- private static final Consumer<ResourceID> NOOP_HEARTBEAT_JOBMANAGER_CONSUMER = ignored -> {};
+ private static final BiConsumer<ResourceID, AllocatedSlotReport> NOOP_HEARTBEAT_JOBMANAGER_CONSUMER = (ignoredA, ignoredB) -> {};
private static final BiConsumer<JobID, Throwable> NOOP_DISCONNECT_JOBMANAGER_CONSUMER = (ignoredA, ignoredB) -> {};
private static final BiFunction<TaskDeploymentDescriptor, JobMasterId, CompletableFuture<Acknowledge>> NOOP_SUBMIT_TASK_CONSUMER = (ignoredA, ignoredB) -> CompletableFuture.completedFuture(Acknowledge.get());
private static final Function<Tuple5<SlotID, JobID, AllocationID, String, ResourceManagerId>, CompletableFuture<Acknowledge>> NOOP_REQUEST_SLOT_FUNCTION = ignored -> CompletableFuture.completedFuture(Acknowledge.get());
@@ -52,7 +53,7 @@ public class TestingTaskExecutorGatewayBuilder {
private String address = "foobar:1234";
private String hostname = "foobar";
- private Consumer<ResourceID> heartbeatJobManagerConsumer = NOOP_HEARTBEAT_JOBMANAGER_CONSUMER;
+ private BiConsumer<ResourceID, AllocatedSlotReport> heartbeatJobManagerConsumer = NOOP_HEARTBEAT_JOBMANAGER_CONSUMER;
private BiConsumer<JobID, Throwable> disconnectJobManagerConsumer = NOOP_DISCONNECT_JOBMANAGER_CONSUMER;
private BiFunction<TaskDeploymentDescriptor, JobMasterId, CompletableFuture<Acknowledge>> submitTaskConsumer = NOOP_SUBMIT_TASK_CONSUMER;
private Function<Tuple5<SlotID, JobID, AllocationID, String, ResourceManagerId>, CompletableFuture<Acknowledge>> requestSlotFunction = NOOP_REQUEST_SLOT_FUNCTION;
@@ -72,7 +73,7 @@ public class TestingTaskExecutorGatewayBuilder {
return this;
}
- public TestingTaskExecutorGatewayBuilder setHeartbeatJobManagerConsumer(Consumer<ResourceID> heartbeatJobManagerConsumer) {
+ public TestingTaskExecutorGatewayBuilder setHeartbeatJobManagerConsumer(BiConsumer<ResourceID, AllocatedSlotReport> heartbeatJobManagerConsumer) {
this.heartbeatJobManagerConsumer = heartbeatJobManagerConsumer;
return this;
}