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 2016/10/21 12:22:07 UTC
[27/50] [abbrv] flink git commit: [FLINK-4347][FLINK-4348] simplify
SlotManager and integrate it with ResourceManager
[FLINK-4347][FLINK-4348] simplify SlotManager and integrate it with ResourceManager
Instead of relying on a full synchronization of all slots information on
every heartbeat, the SlotManager is now responsible for updating its
state. It initially syncs all slots upon registration of the
TaskExecutor. After that, it only receives notifications from the
TaskExecutor when slots become available again. This simplifies the
logic of the SlotManager and makes the slot allocation more predictable
in case of message loss.
Additional changes:
- Move the slot registration and allocation report to the registration
of the TaskExecutor
- Let the TaskExecutor immediately notify the ResourceManager once a
slot becomes free. The ResourceManager has to confirm this
notification. Otherwise, the slot will be blocked because the
ResourceManager's state is not in sync.
- Integrate with handleSlotRequestFailedAtTaskManager and introduce
fencing to protect against TaskExecutors which are not registered
anymore.
- introduce RPC call to notify ResourceManager about free slots
- ignore out-of-date slot requests from ResourceManager at TaskExecutor
- let the ResourceManager update its state instead of relying on heartbeats
- provide ResourceManagerServices to SlotManager
- introduce factory for SlotManager
- keep task gateways and worker information in ResourceManager and
inform SlotManager
- add TaskExecutor test to ensure that a free slot which hasn't been
confirmed by the task executor is correctly blacklisted as long as the
ResourceManager has not confirmed the allocation removal.
- adapt tests
- update javadocs
This closes #2571.
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/0cac6614
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/0cac6614
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/0cac6614
Branch: refs/heads/flip-6
Commit: 0cac6614cc02126a69a87be67b3eaee479e06b51
Parents: 86d5f40
Author: Maximilian Michels <mx...@apache.org>
Authored: Thu Sep 29 15:08:32 2016 +0200
Committer: Till Rohrmann <tr...@apache.org>
Committed: Thu Oct 20 19:47:56 2016 +0200
----------------------------------------------------------------------
.../clusterframework/types/ResourceSlot.java | 12 +-
.../resourcemanager/ResourceManager.java | 115 +++++--
.../resourcemanager/ResourceManagerGateway.java | 31 +-
.../ResourceManagerServices.java | 2 +-
.../resourcemanager/SlotRequestRegistered.java | 33 --
.../resourcemanager/SlotRequestRejected.java | 34 --
.../resourcemanager/SlotRequestReply.java | 41 ---
.../StandaloneResourceManager.java | 8 +-
.../jobmanager/RMSlotRequestRegistered.java | 33 ++
.../jobmanager/RMSlotRequestRejected.java | 34 ++
.../messages/jobmanager/RMSlotRequestReply.java | 41 +++
.../taskexecutor/SlotAvailableReply.java | 47 +++
.../taskexecutor/TMSlotRequestRegistered.java | 35 ++
.../taskexecutor/TMSlotRequestRejected.java | 35 ++
.../taskexecutor/TMSlotRequestReply.java | 58 ++++
.../registration/TaskExecutorRegistration.java | 12 +-
.../registration/WorkerRegistration.java | 42 +++
.../slotmanager/SimpleSlotManager.java | 53 ---
.../slotmanager/SlotManager.java | 326 +++++++------------
.../slotmanager/SlotManagerFactory.java | 31 ++
.../flink/runtime/taskexecutor/SlotReport.java | 19 +-
.../runtime/taskexecutor/TaskExecutor.java | 37 ++-
.../taskexecutor/TaskExecutorGateway.java | 7 +-
...TaskExecutorToResourceManagerConnection.java | 2 +-
.../resourcemanager/ResourceManagerHATest.java | 12 +-
.../ResourceManagerJobMasterTest.java | 4 +-
.../ResourceManagerTaskExecutorTest.java | 53 +--
.../resourcemanager/TestingResourceManager.java | 53 +++
.../resourcemanager/TestingSlotManager.java | 78 +++++
.../slotmanager/SlotManagerTest.java | 239 ++++++--------
.../slotmanager/SlotProtocolTest.java | 92 ++++--
.../runtime/taskexecutor/TaskExecutorTest.java | 96 +++++-
32 files changed, 1087 insertions(+), 628 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/0cac6614/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/types/ResourceSlot.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/types/ResourceSlot.java b/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/types/ResourceSlot.java
index 4a91a79..0b9367d 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/types/ResourceSlot.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/clusterframework/types/ResourceSlot.java
@@ -18,7 +18,7 @@
package org.apache.flink.runtime.clusterframework.types;
-import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
+import org.apache.flink.runtime.resourcemanager.registration.TaskExecutorRegistration;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -35,12 +35,12 @@ public class ResourceSlot implements ResourceIDRetrievable {
private final ResourceProfile resourceProfile;
/** Gateway to the TaskExecutor which owns the slot */
- private final TaskExecutorGateway taskExecutorGateway;
+ private final TaskExecutorRegistration taskExecutorRegistration;
- public ResourceSlot(SlotID slotId, ResourceProfile resourceProfile, TaskExecutorGateway taskExecutorGateway) {
+ public ResourceSlot(SlotID slotId, ResourceProfile resourceProfile, TaskExecutorRegistration taskExecutorRegistration) {
this.slotId = checkNotNull(slotId);
this.resourceProfile = checkNotNull(resourceProfile);
- this.taskExecutorGateway = taskExecutorGateway;
+ this.taskExecutorRegistration = checkNotNull(taskExecutorRegistration);
}
@Override
@@ -56,8 +56,8 @@ public class ResourceSlot implements ResourceIDRetrievable {
return resourceProfile;
}
- public TaskExecutorGateway getTaskExecutorGateway() {
- return taskExecutorGateway;
+ public TaskExecutorRegistration getTaskExecutorRegistration() {
+ return taskExecutorRegistration;
}
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/0cac6614/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManager.java
----------------------------------------------------------------------
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 f45afa3..d2d00cf 100644
--- 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
@@ -24,8 +24,8 @@ import org.apache.flink.api.common.time.Time;
import org.apache.flink.runtime.clusterframework.ApplicationStatus;
import org.apache.flink.runtime.clusterframework.messages.InfoMessage;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
-import org.apache.flink.runtime.clusterframework.types.ResourceIDRetrievable;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
+import org.apache.flink.runtime.clusterframework.types.SlotID;
import org.apache.flink.runtime.concurrent.AcceptFunction;
import org.apache.flink.runtime.concurrent.ApplyFunction;
import org.apache.flink.runtime.concurrent.BiFunction;
@@ -37,7 +37,12 @@ import org.apache.flink.runtime.leaderelection.LeaderContender;
import org.apache.flink.runtime.leaderelection.LeaderElectionService;
import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalListener;
import org.apache.flink.runtime.leaderretrieval.LeaderRetrievalService;
+import org.apache.flink.runtime.resourcemanager.messages.jobmanager.RMSlotRequestRejected;
+import org.apache.flink.runtime.resourcemanager.messages.jobmanager.RMSlotRequestReply;
+import org.apache.flink.runtime.resourcemanager.messages.taskexecutor.SlotAvailableReply;
+import org.apache.flink.runtime.resourcemanager.registration.WorkerRegistration;
import org.apache.flink.runtime.resourcemanager.slotmanager.SlotManager;
+import org.apache.flink.runtime.resourcemanager.slotmanager.SlotManagerFactory;
import org.apache.flink.runtime.rpc.RpcEndpoint;
import org.apache.flink.runtime.rpc.RpcMethod;
import org.apache.flink.runtime.rpc.RpcService;
@@ -45,12 +50,14 @@ import org.apache.flink.runtime.jobmaster.JobMaster;
import org.apache.flink.runtime.jobmaster.JobMasterGateway;
import org.apache.flink.runtime.registration.RegistrationResponse;
+import org.apache.flink.runtime.taskexecutor.SlotReport;
import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
import org.apache.flink.runtime.taskexecutor.TaskExecutorRegistrationSuccess;
import org.apache.flink.runtime.util.LeaderConnectionInfo;
import org.apache.flink.runtime.util.LeaderRetrievalUtils;
import scala.concurrent.duration.FiniteDuration;
+import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
@@ -70,41 +77,54 @@ import static org.apache.flink.util.Preconditions.checkNotNull;
* <li>{@link #requestSlot(UUID, UUID, SlotRequest)} requests a slot from the resource manager</li>
* </ul>
*/
-public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable>
+public abstract class ResourceManager<WorkerType extends Serializable>
extends RpcEndpoint<ResourceManagerGateway>
implements LeaderContender {
- /** The exit code with which the process is stopped in case of a fatal error */
+ /** The exit code with which the process is stopped in case of a fatal error. */
protected static final int EXIT_CODE_FATAL_ERROR = -13;
+ /** All currently registered JobMasterGateways scoped by JobID. */
private final Map<JobID, JobMasterGateway> jobMasterGateways;
+ /** LeaderListeners for all registered JobMasters. */
private final Map<JobID, JobMasterLeaderListener> jobMasterLeaderRetrievalListeners;
- private final Map<ResourceID, WorkerType> taskExecutorGateways;
+ /** All currently registered TaskExecutors with there framework specific worker information. */
+ private final Map<ResourceID, WorkerRegistration<WorkerType>> taskExecutors;
+ /** High availability services for leader retrieval and election. */
private final HighAvailabilityServices highAvailabilityServices;
- private final SlotManager slotManager;
+ /** The factory to construct the SlotManager. */
+ private final SlotManagerFactory slotManagerFactory;
+ /** The SlotManager created by the slotManagerFactory when the ResourceManager is started. */
+ private SlotManager slotManager;
+
+ /** The service to elect a ResourceManager leader. */
private LeaderElectionService leaderElectionService;
+ /** ResourceManager's leader session id which is updated on leader election. */
private UUID leaderSessionID;
+ /** All registered listeners for status updates of the ResourceManager. */
private Map<String, InfoMessageListenerRpcGateway> infoMessageListeners;
+ /** Default timeout for messages */
private final Time timeout = Time.seconds(5);
public ResourceManager(
RpcService rpcService,
HighAvailabilityServices highAvailabilityServices,
- SlotManager slotManager) {
+ SlotManagerFactory slotManagerFactory) {
super(rpcService);
this.highAvailabilityServices = checkNotNull(highAvailabilityServices);
+ this.slotManagerFactory = checkNotNull(slotManagerFactory);
this.jobMasterGateways = new HashMap<>();
- this.slotManager = checkNotNull(slotManager);
this.jobMasterLeaderRetrievalListeners = new HashMap<>();
- this.taskExecutorGateways = new HashMap<>();
+ this.taskExecutors = new HashMap<>();
+ this.leaderSessionID = new UUID(0, 0);
infoMessageListeners = new HashMap<>();
}
@@ -113,9 +133,10 @@ public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable>
// start a leader
try {
super.start();
+ // SlotManager should start first
+ slotManager = slotManagerFactory.create(createResourceManagerServices());
leaderElectionService = highAvailabilityServices.getResourceManagerLeaderElectionService();
leaderElectionService.start(this);
- slotManager.setupResourceManagerServices(new DefaultResourceManagerServices());
// framework specific initialization
initialize();
} catch (Throwable e) {
@@ -196,7 +217,7 @@ public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable>
.handleAsync(new BiFunction<JobMasterGateway, Throwable, RegistrationResponse>() {
@Override
public RegistrationResponse apply(JobMasterGateway jobMasterGateway, Throwable throwable) {
-
+
if (throwable != null) {
return new RegistrationResponse.Decline(throwable.getMessage());
} else {
@@ -234,7 +255,8 @@ public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable>
public Future<RegistrationResponse> registerTaskExecutor(
final UUID resourceManagerLeaderId,
final String taskExecutorAddress,
- final ResourceID resourceID) {
+ final ResourceID resourceID,
+ final SlotReport slotReport) {
return getRpcService().execute(new Callable<TaskExecutorGateway>() {
@Override
@@ -245,7 +267,8 @@ public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable>
resourceID, taskExecutorAddress, leaderSessionID, resourceManagerLeaderId);
throw new Exception("Invalid leader session id");
}
- return getRpcService().connect(taskExecutorAddress, TaskExecutorGateway.class).get(5, TimeUnit.SECONDS);
+ return getRpcService().connect(taskExecutorAddress, TaskExecutorGateway.class)
+ .get(timeout.toMilliseconds(), timeout.getUnit());
}
}).handleAsync(new BiFunction<TaskExecutorGateway, Throwable, RegistrationResponse>() {
@Override
@@ -253,14 +276,17 @@ public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable>
if (throwable != null) {
return new RegistrationResponse.Decline(throwable.getMessage());
} else {
- WorkerType oldWorker = taskExecutorGateways.remove(resourceID);
- if (oldWorker != null) {
+ WorkerRegistration oldRegistration = taskExecutors.remove(resourceID);
+ if (oldRegistration != null) {
// TODO :: suggest old taskExecutor to stop itself
- slotManager.notifyTaskManagerFailure(resourceID);
+ log.info("Replacing old instance of worker for ResourceID {}", resourceID);
}
WorkerType newWorker = workerStarted(resourceID);
- taskExecutorGateways.put(resourceID, newWorker);
- return new TaskExecutorRegistrationSuccess(new InstanceID(), 5000);
+ WorkerRegistration<WorkerType> registration =
+ new WorkerRegistration<>(taskExecutorGateway, newWorker);
+ taskExecutors.put(resourceID, registration);
+ slotManager.registerTaskExecutor(resourceID, registration, slotReport);
+ return new TaskExecutorRegistrationSuccess(registration.getInstanceID(), 5000);
}
}
}, getMainThreadExecutor());
@@ -273,7 +299,7 @@ public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable>
* @return Slot assignment
*/
@RpcMethod
- public SlotRequestReply requestSlot(
+ public RMSlotRequestReply requestSlot(
UUID jobMasterLeaderID,
UUID resourceManagerLeaderID,
SlotRequest slotRequest) {
@@ -290,8 +316,41 @@ public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable>
return slotManager.requestSlot(slotRequest);
} else {
log.info("Ignoring slot request for unknown JobMaster with JobID {}", jobId);
- return new SlotRequestRejected(slotRequest.getAllocationId());
+ return new RMSlotRequestRejected(slotRequest.getAllocationId());
+ }
+ }
+
+ /**
+ * Notification from a TaskExecutor that a slot has become available
+ * @param resourceManagerLeaderId TaskExecutor's resource manager leader id
+ * @param resourceID TaskExecutor's resource id
+ * @param instanceID TaskExecutor's instance id
+ * @param slotID The slot id of the available slot
+ * @return SlotAvailableReply
+ */
+ @RpcMethod
+ public SlotAvailableReply notifySlotAvailable(
+ final UUID resourceManagerLeaderId,
+ final ResourceID resourceID,
+ final InstanceID instanceID,
+ final SlotID slotID) {
+
+ if (resourceManagerLeaderId.equals(leaderSessionID)) {
+ WorkerRegistration<WorkerType> registration = taskExecutors.get(resourceID);
+ if (registration != null) {
+ InstanceID registrationInstanceID = registration.getInstanceID();
+ if (registrationInstanceID.equals(instanceID)) {
+ runAsync(new Runnable() {
+ @Override
+ public void run() {
+ slotManager.notifySlotAvailable(resourceID, slotID);
+ }
+ });
+ return new SlotAvailableReply(leaderSessionID, slotID);
+ }
+ }
}
+ return null;
}
@@ -329,9 +388,9 @@ public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable>
public void run() {
log.info("ResourceManager {} was revoked leadership.", getAddress());
jobMasterGateways.clear();
- taskExecutorGateways.clear();
+ taskExecutors.clear();
slotManager.clearState();
- leaderSessionID = null;
+ leaderSessionID = new UUID(0, 0);
}
});
}
@@ -411,7 +470,7 @@ public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable>
runAsync(new Runnable() {
@Override
public void run() {
- WorkerType worker = taskExecutorGateways.remove(resourceID);
+ WorkerType worker = taskExecutors.remove(resourceID).getWorker();
if (worker != null) {
// TODO :: suggest failed task executor to stop itself
slotManager.notifyTaskManagerFailure(resourceID);
@@ -426,7 +485,7 @@ public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable>
* @return The number of currently started TaskManagers.
*/
public int getNumberOfStartedTaskManagers() {
- return taskExecutorGateways.size();
+ return taskExecutors.size();
}
/**
@@ -507,6 +566,14 @@ public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable>
});
}
+ // ------------------------------------------------------------------------
+ // Resource Manager Services
+ // ------------------------------------------------------------------------
+
+ protected ResourceManagerServices createResourceManagerServices() {
+ return new DefaultResourceManagerServices();
+ }
+
private class DefaultResourceManagerServices implements ResourceManagerServices {
@Override
@@ -520,7 +587,7 @@ public abstract class ResourceManager<WorkerType extends ResourceIDRetrievable>
}
@Override
- public Executor getExecutor() {
+ public Executor getMainThreadExecutor() {
return ResourceManager.this.getMainThreadExecutor();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/0cac6614/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerGateway.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerGateway.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerGateway.java
index 87303a1..3c81227 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerGateway.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerGateway.java
@@ -22,11 +22,16 @@ import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.runtime.clusterframework.ApplicationStatus;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.clusterframework.types.SlotID;
import org.apache.flink.runtime.concurrent.Future;
+import org.apache.flink.runtime.instance.InstanceID;
+import org.apache.flink.runtime.resourcemanager.messages.jobmanager.RMSlotRequestReply;
+import org.apache.flink.runtime.resourcemanager.messages.taskexecutor.SlotAvailableReply;
import org.apache.flink.runtime.rpc.RpcGateway;
import org.apache.flink.runtime.rpc.RpcTimeout;
import org.apache.flink.runtime.jobmaster.JobMaster;
import org.apache.flink.runtime.registration.RegistrationResponse;
+import org.apache.flink.runtime.taskexecutor.SlotReport;
import java.util.UUID;
@@ -56,10 +61,12 @@ public interface ResourceManagerGateway extends RpcGateway {
/**
* Requests a slot from the resource manager.
*
- * @param slotRequest Slot request
- * @return Future slot assignment
+ * @param jobMasterLeaderID leader id of the JobMaster
+ * @param resourceManagerLeaderID leader if of the ResourceMaster
+ * @param slotRequest The slot to request
+ * @return The confirmation that the slot gets allocated
*/
- Future<SlotRequestReply> requestSlot(
+ Future<RMSlotRequestReply> requestSlot(
UUID jobMasterLeaderID,
UUID resourceManagerLeaderID,
SlotRequest slotRequest,
@@ -71,6 +78,7 @@ public interface ResourceManagerGateway extends RpcGateway {
* @param resourceManagerLeaderId The fencing token for the ResourceManager leader
* @param taskExecutorAddress The address of the TaskExecutor that registers
* @param resourceID The resource ID of the TaskExecutor that registers
+ * @param slotReport The slot report containing free and allocated task slots
* @param timeout The timeout for the response.
*
* @return The future to the response by the ResourceManager.
@@ -79,6 +87,23 @@ public interface ResourceManagerGateway extends RpcGateway {
UUID resourceManagerLeaderId,
String taskExecutorAddress,
ResourceID resourceID,
+ SlotReport slotReport,
+ @RpcTimeout Time timeout);
+
+ /**
+ * Sent by the TaskExecutor to notify the ResourceManager that a slot has become available.
+ *
+ * @param resourceManagerLeaderId The ResourceManager leader id
+ * @param resourceID The ResourceID of the TaskExecutor
+ * @param instanceID The InstanceID of the TaskExecutor
+ * @param slotID The SlotID of the freed slot
+ * @return The confirmation by the ResourceManager
+ */
+ Future<SlotAvailableReply> notifySlotAvailable(
+ UUID resourceManagerLeaderId,
+ ResourceID resourceID,
+ InstanceID instanceID,
+ SlotID slotID,
@RpcTimeout Time timeout);
/**
http://git-wip-us.apache.org/repos/asf/flink/blob/0cac6614/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerServices.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerServices.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerServices.java
index 30994dc..b997a3a 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerServices.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/ResourceManagerServices.java
@@ -39,6 +39,6 @@ public interface ResourceManagerServices {
/**
* Gets the executor which executes in the main thread of the ResourceManager
*/
- Executor getExecutor();
+ Executor getMainThreadExecutor();
}
http://git-wip-us.apache.org/repos/asf/flink/blob/0cac6614/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/SlotRequestRegistered.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/SlotRequestRegistered.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/SlotRequestRegistered.java
deleted file mode 100644
index f719dce..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/SlotRequestRegistered.java
+++ /dev/null
@@ -1,33 +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.resourcemanager;
-
-import org.apache.flink.runtime.clusterframework.types.AllocationID;
-
-/**
- * Acknowledgment by the ResourceManager for a SlotRequest from the JobManager
- */
-public class SlotRequestRegistered extends SlotRequestReply {
-
- private static final long serialVersionUID = 4760320859275256855L;
-
- public SlotRequestRegistered(AllocationID allocationID) {
- super(allocationID);
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/0cac6614/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/SlotRequestRejected.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/SlotRequestRejected.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/SlotRequestRejected.java
deleted file mode 100644
index 282a7d5..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/SlotRequestRejected.java
+++ /dev/null
@@ -1,34 +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.resourcemanager;
-
-import org.apache.flink.runtime.clusterframework.types.AllocationID;
-
-/**
- * Rejection message by the ResourceManager for a SlotRequest from the JobManager
- */
-public class SlotRequestRejected extends SlotRequestReply {
-
- private static final long serialVersionUID = 9049346740895325144L;
-
- public SlotRequestRejected(AllocationID allocationID) {
- super(allocationID);
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/0cac6614/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/SlotRequestReply.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/SlotRequestReply.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/SlotRequestReply.java
deleted file mode 100644
index 1b85d0c..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/SlotRequestReply.java
+++ /dev/null
@@ -1,41 +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.resourcemanager;
-
-import org.apache.flink.runtime.clusterframework.types.AllocationID;
-
-import java.io.Serializable;
-
-/**
- * Acknowledgment by the ResourceManager for a SlotRequest from the JobManager
- */
-public abstract class SlotRequestReply implements Serializable {
-
- private static final long serialVersionUID = 42;
-
- private final AllocationID allocationID;
-
- public SlotRequestReply(AllocationID allocationID) {
- this.allocationID = allocationID;
- }
-
- public AllocationID getAllocationID() {
- return allocationID;
- }
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/0cac6614/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/StandaloneResourceManager.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/StandaloneResourceManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/StandaloneResourceManager.java
index deca8d3..f9f55f8 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/StandaloneResourceManager.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/StandaloneResourceManager.java
@@ -22,7 +22,7 @@ import org.apache.flink.runtime.clusterframework.ApplicationStatus;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
import org.apache.flink.runtime.clusterframework.types.ResourceProfile;
import org.apache.flink.runtime.highavailability.HighAvailabilityServices;
-import org.apache.flink.runtime.resourcemanager.slotmanager.SlotManager;
+import org.apache.flink.runtime.resourcemanager.slotmanager.SlotManagerFactory;
import org.apache.flink.runtime.rpc.RpcService;
/**
@@ -34,9 +34,9 @@ import org.apache.flink.runtime.rpc.RpcService;
public class StandaloneResourceManager extends ResourceManager<ResourceID> {
public StandaloneResourceManager(RpcService rpcService,
- HighAvailabilityServices highAvailabilityServices,
- SlotManager slotManager) {
- super(rpcService, highAvailabilityServices, slotManager);
+ HighAvailabilityServices highAvailabilityServices,
+ SlotManagerFactory slotManagerFactory) {
+ super(rpcService, highAvailabilityServices, slotManagerFactory);
}
@Override
http://git-wip-us.apache.org/repos/asf/flink/blob/0cac6614/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/messages/jobmanager/RMSlotRequestRegistered.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/messages/jobmanager/RMSlotRequestRegistered.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/messages/jobmanager/RMSlotRequestRegistered.java
new file mode 100644
index 0000000..01bc532
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/messages/jobmanager/RMSlotRequestRegistered.java
@@ -0,0 +1,33 @@
+/*
+ * 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.resourcemanager.messages.jobmanager;
+
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+
+/**
+ * Acknowledgment by the ResourceManager for a SlotRequest from the JobManager
+ */
+public class RMSlotRequestRegistered extends RMSlotRequestReply {
+
+ private static final long serialVersionUID = 4760320859275256855L;
+
+ public RMSlotRequestRegistered(AllocationID allocationID) {
+ super(allocationID);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/0cac6614/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/messages/jobmanager/RMSlotRequestRejected.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/messages/jobmanager/RMSlotRequestRejected.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/messages/jobmanager/RMSlotRequestRejected.java
new file mode 100644
index 0000000..649d61c
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/messages/jobmanager/RMSlotRequestRejected.java
@@ -0,0 +1,34 @@
+/*
+ * 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.resourcemanager.messages.jobmanager;
+
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+
+/**
+ * Rejection message by the ResourceManager for a SlotRequest from the JobManager
+ */
+public class RMSlotRequestRejected extends RMSlotRequestReply {
+
+ private static final long serialVersionUID = 9049346740895325144L;
+
+ public RMSlotRequestRejected(AllocationID allocationID) {
+ super(allocationID);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/0cac6614/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/messages/jobmanager/RMSlotRequestReply.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/messages/jobmanager/RMSlotRequestReply.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/messages/jobmanager/RMSlotRequestReply.java
new file mode 100644
index 0000000..66e1911
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/messages/jobmanager/RMSlotRequestReply.java
@@ -0,0 +1,41 @@
+/*
+ * 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.resourcemanager.messages.jobmanager;
+
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+
+import java.io.Serializable;
+
+/**
+ * Acknowledgment by the ResourceManager for a SlotRequest from the JobManager
+ */
+public abstract class RMSlotRequestReply implements Serializable {
+
+ private static final long serialVersionUID = 42;
+
+ private final AllocationID allocationID;
+
+ public RMSlotRequestReply(AllocationID allocationID) {
+ this.allocationID = allocationID;
+ }
+
+ public AllocationID getAllocationID() {
+ return allocationID;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/0cac6614/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/messages/taskexecutor/SlotAvailableReply.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/messages/taskexecutor/SlotAvailableReply.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/messages/taskexecutor/SlotAvailableReply.java
new file mode 100644
index 0000000..f2e0105
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/messages/taskexecutor/SlotAvailableReply.java
@@ -0,0 +1,47 @@
+/*
+ * 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.resourcemanager.messages.taskexecutor;
+
+import org.apache.flink.runtime.clusterframework.types.SlotID;
+
+import java.io.Serializable;
+import java.util.UUID;
+
+/**
+ * Sent by the ResourceManager to the TaskExecutor confirm receipt of
+ * {@code org.apache.flink.runtime.resourcemanager.ResourceManagerGateway.notifySlotAvailable}.
+ */
+public class SlotAvailableReply implements Serializable {
+
+ private final UUID resourceManagerLeaderID;
+
+ private final SlotID slotID;
+
+ public SlotAvailableReply(UUID resourceManagerLeaderID, SlotID slotID) {
+ this.resourceManagerLeaderID = resourceManagerLeaderID;
+ this.slotID = slotID;
+ }
+
+ public UUID getResourceManagerLeaderID() {
+ return resourceManagerLeaderID;
+ }
+
+ public SlotID getSlotID() {
+ return slotID;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/0cac6614/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/messages/taskexecutor/TMSlotRequestRegistered.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/messages/taskexecutor/TMSlotRequestRegistered.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/messages/taskexecutor/TMSlotRequestRegistered.java
new file mode 100644
index 0000000..c0f0f49
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/messages/taskexecutor/TMSlotRequestRegistered.java
@@ -0,0 +1,35 @@
+/*
+ * 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.resourcemanager.messages.taskexecutor;
+
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.instance.InstanceID;
+
+/**
+ * Acknowledgment by the TaskExecutor for a SlotRequest from the ResourceManager
+ */
+public class TMSlotRequestRegistered extends TMSlotRequestReply {
+
+ private static final long serialVersionUID = 4760320859275256855L;
+
+ public TMSlotRequestRegistered(InstanceID instanceID, ResourceID resourceID, AllocationID allocationID) {
+ super(instanceID, resourceID, allocationID);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/0cac6614/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/messages/taskexecutor/TMSlotRequestRejected.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/messages/taskexecutor/TMSlotRequestRejected.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/messages/taskexecutor/TMSlotRequestRejected.java
new file mode 100644
index 0000000..9b10a35
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/messages/taskexecutor/TMSlotRequestRejected.java
@@ -0,0 +1,35 @@
+/*
+ * 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.resourcemanager.messages.taskexecutor;
+
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.instance.InstanceID;
+
+/**
+ * Rejection by the TaskExecutor for a SlotRequest from the ResourceManager
+ */
+public class TMSlotRequestRejected extends TMSlotRequestReply {
+
+ private static final long serialVersionUID = 9049346740895325144L;
+
+ public TMSlotRequestRejected(InstanceID instanceID, ResourceID resourceID, AllocationID allocationID) {
+ super(instanceID, resourceID, allocationID);
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/0cac6614/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/messages/taskexecutor/TMSlotRequestReply.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/messages/taskexecutor/TMSlotRequestReply.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/messages/taskexecutor/TMSlotRequestReply.java
new file mode 100644
index 0000000..b23b6e9
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/messages/taskexecutor/TMSlotRequestReply.java
@@ -0,0 +1,58 @@
+/*
+ * 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.resourcemanager.messages.taskexecutor;
+
+import org.apache.flink.runtime.clusterframework.types.AllocationID;
+import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.instance.InstanceID;
+
+import java.io.Serializable;
+
+/**
+ * Acknowledgment by the TaskExecutor for a SlotRequest from the ResourceManager
+ */
+public abstract class TMSlotRequestReply implements Serializable {
+
+ private static final long serialVersionUID = 42;
+
+ private final InstanceID instanceID;
+
+ private final ResourceID resourceID;
+
+ private final AllocationID allocationID;
+
+ protected TMSlotRequestReply(InstanceID instanceID, ResourceID resourceID, AllocationID allocationID) {
+ this.instanceID = instanceID;
+ this.resourceID = resourceID;
+ this.allocationID = allocationID;
+ }
+
+ public InstanceID getInstanceID() {
+ return instanceID;
+ }
+
+ public ResourceID getResourceID() {
+ return resourceID;
+ }
+
+ public AllocationID getAllocationID() {
+ return allocationID;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/0cac6614/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/registration/TaskExecutorRegistration.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/registration/TaskExecutorRegistration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/registration/TaskExecutorRegistration.java
index 6b21f5c..bfa9c00 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/registration/TaskExecutorRegistration.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/registration/TaskExecutorRegistration.java
@@ -24,20 +24,20 @@ import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
import java.io.Serializable;
/**
- * This class is responsible for group the TaskExecutorGateway and the InstanceID of a registered task executor.
+ * This class is responsible for grouping the TaskExecutorGateway and the InstanceID
+ * of a registered task executor.
*/
public class TaskExecutorRegistration implements Serializable {
private static final long serialVersionUID = -2062957799469434614L;
- private TaskExecutorGateway taskExecutorGateway;
+ private final InstanceID instanceID;
- private InstanceID instanceID;
+ private TaskExecutorGateway taskExecutorGateway;
- public TaskExecutorRegistration(TaskExecutorGateway taskExecutorGateway,
- InstanceID instanceID) {
+ public TaskExecutorRegistration(TaskExecutorGateway taskExecutorGateway) {
+ this.instanceID = new InstanceID();
this.taskExecutorGateway = taskExecutorGateway;
- this.instanceID = instanceID;
}
public InstanceID getInstanceID() {
http://git-wip-us.apache.org/repos/asf/flink/blob/0cac6614/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/registration/WorkerRegistration.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/registration/WorkerRegistration.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/registration/WorkerRegistration.java
new file mode 100644
index 0000000..ff28f94
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/registration/WorkerRegistration.java
@@ -0,0 +1,42 @@
+/*
+ * 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.resourcemanager.registration;
+
+import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
+
+import java.io.Serializable;
+
+/**
+ * This class extends the {@link TaskExecutorRegistration}, adding the worker information.
+ */
+public class WorkerRegistration<WorkerType extends Serializable> extends TaskExecutorRegistration {
+
+ private static final long serialVersionUID = -2062957799469434614L;
+
+ private WorkerType worker;
+
+ public WorkerRegistration(TaskExecutorGateway taskExecutorGateway, WorkerType worker) {
+ super(taskExecutorGateway);
+ this.worker = worker;
+ }
+
+ public WorkerType getWorker() {
+ return worker;
+ }
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/0cac6614/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SimpleSlotManager.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SimpleSlotManager.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SimpleSlotManager.java
deleted file mode 100644
index ae1de5a..0000000
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SimpleSlotManager.java
+++ /dev/null
@@ -1,53 +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.resourcemanager.slotmanager;
-
-import org.apache.flink.runtime.clusterframework.types.AllocationID;
-import org.apache.flink.runtime.clusterframework.types.ResourceSlot;
-import org.apache.flink.runtime.clusterframework.types.SlotID;
-import org.apache.flink.runtime.resourcemanager.SlotRequest;
-
-import java.util.Iterator;
-import java.util.Map;
-
-/**
- * A simple SlotManager which ignores resource profiles.
- */
-public class SimpleSlotManager extends SlotManager {
-
- @Override
- protected ResourceSlot chooseSlotToUse(SlotRequest request, Map<SlotID, ResourceSlot> freeSlots) {
- final Iterator<ResourceSlot> slotIterator = freeSlots.values().iterator();
- if (slotIterator.hasNext()) {
- return slotIterator.next();
- } else {
- return null;
- }
- }
-
- @Override
- protected SlotRequest chooseRequestToFulfill(ResourceSlot offeredSlot, Map<AllocationID, SlotRequest> pendingRequests) {
- final Iterator<SlotRequest> requestIterator = pendingRequests.values().iterator();
- if (requestIterator.hasNext()) {
- return requestIterator.next();
- } else {
- return null;
- }
- }
-
-}
http://git-wip-us.apache.org/repos/asf/flink/blob/0cac6614/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManager.java
----------------------------------------------------------------------
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 a56b2f6..7eb2d78 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
@@ -28,11 +28,12 @@ import org.apache.flink.runtime.concurrent.BiFunction;
import org.apache.flink.runtime.concurrent.Future;
import org.apache.flink.runtime.resourcemanager.ResourceManagerServices;
import org.apache.flink.runtime.resourcemanager.SlotRequest;
-import org.apache.flink.runtime.resourcemanager.SlotRequestRegistered;
-import org.apache.flink.runtime.resourcemanager.SlotRequestReply;
+import org.apache.flink.runtime.resourcemanager.messages.jobmanager.RMSlotRequestRegistered;
+import org.apache.flink.runtime.resourcemanager.messages.taskexecutor.TMSlotRequestRejected;
+import org.apache.flink.runtime.resourcemanager.messages.taskexecutor.TMSlotRequestReply;
+import org.apache.flink.runtime.resourcemanager.registration.TaskExecutorRegistration;
import org.apache.flink.runtime.taskexecutor.SlotReport;
import org.apache.flink.runtime.taskexecutor.SlotStatus;
-import org.apache.flink.runtime.taskexecutor.TaskExecutorGateway;
import org.apache.flink.util.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -64,8 +65,11 @@ public abstract class SlotManager {
protected final Logger LOG = LoggerFactory.getLogger(getClass());
+ /** The Resource allocation provider */
+ protected final ResourceManagerServices rmServices;
+
/** All registered task managers with ResourceID and gateway. */
- private final Map<ResourceID, TaskExecutorGateway> taskManagerGateways;
+ private final Map<ResourceID, TaskExecutorRegistration> taskManagers;
/** All registered slots, including free and allocated slots */
private final Map<ResourceID, Map<SlotID, ResourceSlot>> registeredSlots;
@@ -84,29 +88,17 @@ public abstract class SlotManager {
/** The current leader id set by the ResourceManager */
private UUID leaderID;
- /** The Resource allocation provider */
- private ResourceManagerServices resourceManagerServices;
-
- public SlotManager() {
+ public SlotManager(ResourceManagerServices rmServices) {
+ this.rmServices = checkNotNull(rmServices);
this.registeredSlots = new HashMap<>(16);
this.pendingSlotRequests = new LinkedHashMap<>(16);
this.freeSlots = new HashMap<>(16);
this.allocationMap = new AllocationMap();
- this.taskManagerGateways = new HashMap<>();
+ this.taskManagers = new HashMap<>();
this.timeout = Time.seconds(10);
+ this.leaderID = new UUID(0, 0);
}
- /**
- * Initializes the resource supplier which is needed to request new resources.
- */
- public void setupResourceManagerServices(ResourceManagerServices resourceManagerServices) {
- if (this.resourceManagerServices != null) {
- throw new IllegalStateException("ResourceManagerServices may only be set once.");
- }
- this.resourceManagerServices = resourceManagerServices;
- }
-
-
// ------------------------------------------------------------------------
// slot managements
// ------------------------------------------------------------------------
@@ -118,13 +110,13 @@ public abstract class SlotManager {
* RPC's main thread to avoid race condition).
*
* @param request The detailed request of the slot
- * @return SlotRequestRegistered The confirmation message to be send to the caller
+ * @return RMSlotRequestRegistered The confirmation message to be send to the caller
*/
- public SlotRequestRegistered requestSlot(final SlotRequest request) {
+ public RMSlotRequestRegistered requestSlot(final SlotRequest request) {
final AllocationID allocationId = request.getAllocationId();
if (isRequestDuplicated(request)) {
LOG.warn("Duplicated slot request, AllocationID:{}", allocationId);
- return null;
+ return new RMSlotRequestRegistered(allocationId);
}
// try to fulfil the request with current free slots
@@ -136,53 +128,38 @@ public abstract class SlotManager {
// record this allocation in bookkeeping
allocationMap.addAllocation(slot.getSlotId(), allocationId);
// remove selected slot from free pool
- final ResourceSlot removedSlot = freeSlots.remove(slot.getSlotId());
-
- final Future<SlotRequestReply> slotRequestReplyFuture =
- slot.getTaskExecutorGateway().requestSlot(allocationId, leaderID, timeout);
-
- slotRequestReplyFuture.handleAsync(new BiFunction<SlotRequestReply, Throwable, Object>() {
- @Override
- public Object apply(SlotRequestReply slotRequestReply, Throwable throwable) {
- if (throwable != null) {
- // we failed, put the slot and the request back again
- if (allocationMap.isAllocated(slot.getSlotId())) {
- // only re-add if the slot hasn't been removed in the meantime
- freeSlots.put(slot.getSlotId(), removedSlot);
- }
- pendingSlotRequests.put(allocationId, request);
- }
- return null;
- }
- }, resourceManagerServices.getExecutor());
+ freeSlots.remove(slot.getSlotId());
+
+ sendSlotRequest(slot, request);
} else {
LOG.info("Cannot fulfil slot request, try to allocate a new container for it, " +
"AllocationID:{}, JobID:{}", allocationId, request.getJobId());
- Preconditions.checkState(resourceManagerServices != null,
+ Preconditions.checkState(rmServices != null,
"Attempted to allocate resources but no ResourceManagerServices set.");
- resourceManagerServices.allocateResource(request.getResourceProfile());
+ rmServices.allocateResource(request.getResourceProfile());
pendingSlotRequests.put(allocationId, request);
}
- return new SlotRequestRegistered(allocationId);
+ return new RMSlotRequestRegistered(allocationId);
}
/**
- * Sync slot status with TaskManager's SlotReport.
+ * Notifies the SlotManager that a slot is available again after being allocated.
+ * @param slotID slot id of available slot
*/
- public void updateSlotStatus(final SlotReport slotReport) {
- for (SlotStatus slotStatus : slotReport.getSlotsStatus()) {
- updateSlotStatus(slotStatus);
+ public void notifySlotAvailable(ResourceID resourceID, SlotID slotID) {
+ if (!allocationMap.isAllocated(slotID)) {
+ throw new IllegalStateException("Slot was not previously allocated but " +
+ "TaskManager reports it as available again");
}
- }
-
- /**
- * Registers a TaskExecutor
- * @param resourceID TaskExecutor's ResourceID
- * @param gateway TaskExcutor's gateway
- */
- public void registerTaskExecutor(ResourceID resourceID, TaskExecutorGateway gateway) {
- this.taskManagerGateways.put(resourceID, gateway);
+ allocationMap.removeAllocation(slotID);
+ final Map<SlotID, ResourceSlot> slots = registeredSlots.get(resourceID);
+ ResourceSlot freeSlot = slots.get(slotID);
+ if (freeSlot == null) {
+ throw new IllegalStateException("Slot was not registered with SlotManager but " +
+ "TaskManager reported it to be available.");
+ }
+ handleFreeSlot(freeSlot);
}
/**
@@ -200,51 +177,72 @@ public abstract class SlotManager {
* @param originalRequest The original slot request
* @param slotId The target SlotID
*/
- public void handleSlotRequestFailedAtTaskManager(final SlotRequest originalRequest, final SlotID slotId) {
+ void handleSlotRequestFailedAtTaskManager(final SlotRequest originalRequest, final SlotID slotId) {
final AllocationID originalAllocationId = originalRequest.getAllocationId();
LOG.info("Slot request failed at TaskManager, SlotID:{}, AllocationID:{}, JobID:{}",
slotId, originalAllocationId, originalRequest.getJobId());
- // verify the allocation info before we do anything
- if (freeSlots.containsKey(slotId)) {
- // this slot is currently empty, no need to de-allocate it from our allocations
- LOG.info("Original slot is somehow empty, retrying this request");
+ if (allocationMap.isAllocated(slotId)) {
+ final AllocationID expectedAllocationId = allocationMap.getAllocationID(slotId);
- // before retry, we should double check whether this request was allocated by some other ways
- if (!allocationMap.isAllocated(originalAllocationId)) {
- requestSlot(originalRequest);
+ // check whether we have an agreement on whom this slot belongs to
+ if (originalAllocationId.equals(expectedAllocationId)) {
+ LOG.info("De-allocate this request and retry");
+ allocationMap.removeAllocation(expectedAllocationId);
+ pendingSlotRequests.put(originalRequest.getAllocationId(), originalRequest);
+ ResourceSlot slot = checkNotNull(getRegisteredSlot(slotId));
+ // treat this slot as empty and retry with a different request
+ handleFreeSlot(slot);
} else {
- LOG.info("The failed request has somehow been allocated, SlotID:{}",
- allocationMap.getSlotID(originalAllocationId));
+ LOG.error("Slot request failed for slot {} with allocation id {}:" +
+ " Allocation id did not match the expected allocation id {}.",
+ slotId, originalAllocationId, expectedAllocationId);
}
- } else if (allocationMap.isAllocated(slotId)) {
- final AllocationID currentAllocationId = allocationMap.getAllocationID(slotId);
+ } else {
+ LOG.error("Slot request failed for slot {} with allocation id {}: " +
+ "Slot was not previously registered.",
+ slotId, originalAllocationId);
+ }
+ }
- // check whether we have an agreement on whom this slot belongs to
- if (originalAllocationId.equals(currentAllocationId)) {
- LOG.info("De-allocate this request and retry");
- allocationMap.removeAllocation(currentAllocationId);
+ /**
+ * Registers a TaskExecutor
+ * @param resourceID TaskExecutor's ResourceID
+ * @param registration TaskExecutor's registration
+ * @param slotReport TaskExecutor's free and allocated slots
+ */
+ public void registerTaskExecutor(
+ ResourceID resourceID,
+ TaskExecutorRegistration registration,
+ SlotReport slotReport) {
- // put this slot back to free pool
- ResourceSlot slot = checkNotNull(getRegisteredSlot(slotId));
- freeSlots.put(slotId, slot);
+ if (taskManagers.get(resourceID) != null) {
+ notifyTaskManagerFailure(resourceID);
+ }
- // retry the request
- requestSlot(originalRequest);
- } else {
- // the slot is taken by someone else, no need to de-allocate it from our allocations
- LOG.info("Original slot is taken by someone else, current AllocationID:{}", currentAllocationId);
+ this.taskManagers.put(resourceID, registration);
- // before retry, we should double check whether this request was allocated by some other ways
- if (!allocationMap.isAllocated(originalAllocationId)) {
- requestSlot(originalRequest);
- } else {
- LOG.info("The failed request is somehow been allocated, SlotID:{}",
- allocationMap.getSlotID(originalAllocationId));
- }
+ for (SlotStatus slotStatus : slotReport.getSlotsStatus()) {
+ final SlotID slotId = slotStatus.getSlotID();
+
+ final TaskExecutorRegistration taskExecutorRegistration = taskManagers.get(slotId.getResourceID());
+ if (taskExecutorRegistration == null) {
+ LOG.info("Received SlotStatus but ResourceID {} is unknown to the SlotManager",
+ slotId.getResourceID());
+ return;
+ }
+
+ final ResourceSlot slot = new ResourceSlot(slotId, slotStatus.getProfiler(), taskExecutorRegistration);
+
+ registerNewSlot(slot);
+ LOG.info("New slot appeared, SlotID:{}, AllocationID:{}", slotId, slotStatus.getAllocationID());
+
+ if (slotStatus.getAllocationID() != null) {
+ // slot in use, record this in bookkeeping
+ allocationMap.addAllocation(slotId, slotStatus.getAllocationID());
+ } else {
+ handleFreeSlot(slot);
}
- } else {
- LOG.error("BUG! {} is neither in free pool nor in allocated pool", slotId);
}
}
@@ -255,7 +253,7 @@ public abstract class SlotManager {
*/
public void notifyTaskManagerFailure(final ResourceID resourceId) {
LOG.info("Resource:{} been notified failure", resourceId);
- taskManagerGateways.remove(resourceId);
+ taskManagers.remove(resourceId);
final Map<SlotID, ResourceSlot> slotIdsToRemove = registeredSlots.remove(resourceId);
if (slotIdsToRemove != null) {
for (SlotID slotId : slotIdsToRemove.keySet()) {
@@ -276,92 +274,6 @@ public abstract class SlotManager {
// ------------------------------------------------------------------------
/**
- * Update slot status based on TaskManager's report. There are mainly two situations when we receive the report:
- * <ul>
- * <li>1. The slot is newly registered.</li>
- * <li>2. The slot has registered, it contains its current status.</li>
- * </ul>
- * <p>
- * Regarding 1: It's fairly simple, we just record this slot's status, and trigger schedule if slot is empty.
- * <p>
- * Regarding 2: It will cause some weird situation since we may have some time-gap on how the slot's status really
- * is. We may have some updates on the slot's allocation, but it doesn't reflected by TaskManager's heartbeat yet,
- * and we may make some wrong decision if we cannot guarantee we have the exact status about all the slots. So
- * the principle here is: We always trust TaskManager's heartbeat, we will correct our information based on that
- * and take next action based on the diff between our information and heartbeat status.
- *
- * @param reportedStatus Reported slot status
- */
- void updateSlotStatus(final SlotStatus reportedStatus) {
- final SlotID slotId = reportedStatus.getSlotID();
-
- final TaskExecutorGateway taskExecutorGateway = taskManagerGateways.get(slotId.getResourceID());
- if (taskExecutorGateway == null) {
- LOG.info("Received SlotStatus but ResourceID {} is unknown to the SlotManager",
- slotId.getResourceID());
- return;
- }
-
- final ResourceSlot slot = new ResourceSlot(slotId, reportedStatus.getProfiler(), taskExecutorGateway);
-
- if (registerNewSlot(slot)) {
- // we have a newly registered slot
- LOG.info("New slot appeared, SlotID:{}, AllocationID:{}", slotId, reportedStatus.getAllocationID());
-
- if (reportedStatus.getAllocationID() != null) {
- // slot in use, record this in bookkeeping
- allocationMap.addAllocation(slotId, reportedStatus.getAllocationID());
- } else {
- handleFreeSlot(slot);
- }
- } else {
- // slot exists, update current information
- if (reportedStatus.getAllocationID() != null) {
- // slot is reported in use
- final AllocationID reportedAllocationId = reportedStatus.getAllocationID();
-
- // check whether we also thought this slot is in use
- if (allocationMap.isAllocated(slotId)) {
- // we also think that slot is in use, check whether the AllocationID matches
- final AllocationID currentAllocationId = allocationMap.getAllocationID(slotId);
-
- if (!reportedAllocationId.equals(currentAllocationId)) {
- LOG.info("Slot allocation info mismatch! SlotID:{}, current:{}, reported:{}",
- slotId, currentAllocationId, reportedAllocationId);
-
- // seems we have a disagreement about the slot assignments, need to correct it
- allocationMap.removeAllocation(slotId);
- allocationMap.addAllocation(slotId, reportedAllocationId);
- }
- } else {
- LOG.info("Slot allocation info mismatch! SlotID:{}, current:null, reported:{}",
- slotId, reportedAllocationId);
-
- // we thought the slot is free, should correct this information
- allocationMap.addAllocation(slotId, reportedStatus.getAllocationID());
-
- // remove this slot from free slots pool
- freeSlots.remove(slotId);
- }
- } else {
- // slot is reported empty
-
- // check whether we also thought this slot is empty
- if (allocationMap.isAllocated(slotId)) {
- LOG.info("Slot allocation info mismatch! SlotID:{}, current:{}, reported:null",
- slotId, allocationMap.getAllocationID(slotId));
-
- // we thought the slot is in use, correct it
- allocationMap.removeAllocation(slotId);
-
- // we have a free slot!
- handleFreeSlot(slot);
- }
- }
- }
- }
-
- /**
* When we have a free slot, try to fulfill the pending request first. If any request can be fulfilled,
* record this allocation in bookkeeping and send slot request to TaskManager, else we just add this slot
* to the free pool.
@@ -373,32 +285,45 @@ public abstract class SlotManager {
if (chosenRequest != null) {
final AllocationID allocationId = chosenRequest.getAllocationId();
- final SlotRequest removedSlotRequest = pendingSlotRequests.remove(allocationId);
+ final SlotRequest slotRequest = pendingSlotRequests.remove(allocationId);
LOG.info("Assigning SlotID({}) to AllocationID({}), JobID:{}", freeSlot.getSlotId(),
allocationId, chosenRequest.getJobId());
allocationMap.addAllocation(freeSlot.getSlotId(), allocationId);
- final Future<SlotRequestReply> slotRequestReplyFuture =
- freeSlot.getTaskExecutorGateway().requestSlot(allocationId, leaderID, timeout);
-
- slotRequestReplyFuture.handleAsync(new BiFunction<SlotRequestReply, Throwable, Object>() {
- @Override
- public Object apply(SlotRequestReply slotRequestReply, Throwable throwable) {
- if (throwable != null) {
- // we failed, add the request back again
- if (allocationMap.isAllocated(freeSlot.getSlotId())) {
- pendingSlotRequests.put(allocationId, removedSlotRequest);
- }
- }
- return null;
- }
- }, resourceManagerServices.getExecutor());
+ sendSlotRequest(freeSlot, slotRequest);
} else {
freeSlots.put(freeSlot.getSlotId(), freeSlot);
}
}
+ private void sendSlotRequest(final ResourceSlot freeSlot, final SlotRequest slotRequest) {
+
+ final AllocationID allocationID = slotRequest.getAllocationId();
+ final TaskExecutorRegistration registration = freeSlot.getTaskExecutorRegistration();
+ final Future<TMSlotRequestReply> slotRequestReplyFuture =
+ registration.getTaskExecutorGateway()
+ .requestSlot(freeSlot.getSlotId(), allocationID, leaderID, timeout);
+
+ slotRequestReplyFuture.handleAsync(new BiFunction<TMSlotRequestReply, Throwable, Void>() {
+ @Override
+ public Void apply(TMSlotRequestReply slotRequestReply, Throwable throwable) {
+ TaskExecutorRegistration current = taskManagers.get(slotRequestReply.getResourceID());
+ if (current != null && current.getInstanceID().equals(slotRequestReply.getInstanceID())) {
+ if (throwable != null || slotRequestReply instanceof TMSlotRequestRejected) {
+ handleSlotRequestFailedAtTaskManager(slotRequest, freeSlot.getSlotId());
+ } else {
+ LOG.debug("Successfully registered slot {} ", freeSlot.getSlotId());
+ }
+ } else {
+ LOG.debug("Discarding message from obsolete TaskExecutor with InstanceID {}",
+ slotRequestReply.getInstanceID());
+ }
+ return null;
+ }
+ }, rmServices.getMainThreadExecutor());
+ }
+
/**
* Check whether the request is duplicated. We use AllocationID to identify slot request, for each
* formerly received slot request, it is either in pending list or already been allocated.
@@ -413,18 +338,17 @@ public abstract class SlotManager {
}
/**
- * Try to register slot, and tell if this slot is newly registered.
+ * Registers a new slot with the SlotManager.
*
- * @param slot The ResourceSlot which will be checked and registered
- * @return <tt>true</tt> if we meet a new slot
+ * @param slot The ResourceSlot which will be registered
*/
- private boolean registerNewSlot(final ResourceSlot slot) {
+ private void registerNewSlot(final ResourceSlot slot) {
final SlotID slotId = slot.getSlotId();
final ResourceID resourceId = slotId.getResourceID();
if (!registeredSlots.containsKey(resourceId)) {
registeredSlots.put(resourceId, new HashMap<SlotID, ResourceSlot>());
}
- return registeredSlots.get(resourceId).put(slotId, slot) == null;
+ registeredSlots.get(resourceId).put(slotId, slot);
}
private ResourceSlot getRegisteredSlot(final SlotID slotId) {
@@ -559,12 +483,12 @@ public abstract class SlotManager {
* Clears the state of the SlotManager after leadership revokal
*/
public void clearState() {
- taskManagerGateways.clear();
+ taskManagers.clear();
registeredSlots.clear();
pendingSlotRequests.clear();
freeSlots.clear();
allocationMap.clear();
- leaderID = null;
+ leaderID = new UUID(0, 0);
}
// ------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/0cac6614/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerFactory.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerFactory.java b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerFactory.java
new file mode 100644
index 0000000..b4e9c99
--- /dev/null
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/resourcemanager/slotmanager/SlotManagerFactory.java
@@ -0,0 +1,31 @@
+/*
+ * 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.resourcemanager.slotmanager;
+
+import org.apache.flink.runtime.resourcemanager.ResourceManagerServices;
+
+/**
+ * Factory to create a SlotManager and provide it with dependencies.
+ */
+public interface SlotManagerFactory {
+
+ /**
+ * Creates a SlotManager and provides it with ResourceManager services.
+ */
+ SlotManager create(ResourceManagerServices rmServices);
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/0cac6614/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/SlotReport.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/SlotReport.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/SlotReport.java
index a5de2d5..54adce6 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/SlotReport.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/SlotReport.java
@@ -18,9 +18,8 @@
package org.apache.flink.runtime.taskexecutor;
-import org.apache.flink.runtime.clusterframework.types.ResourceID;
-
import java.io.Serializable;
+import java.util.Collections;
import java.util.List;
import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -37,20 +36,20 @@ public class SlotReport implements Serializable {
/** The slots status of the TaskManager */
private final List<SlotStatus> slotsStatus;
- /** The resource id which identifies the TaskManager */
- private final ResourceID resourceID;
+ public SlotReport() {
+ this(Collections.<SlotStatus>emptyList());
+ }
+
+ public SlotReport(SlotStatus slotStatus) {
+ this(Collections.singletonList(slotStatus));
+ }
- public SlotReport(final List<SlotStatus> slotsStatus, final ResourceID resourceID) {
+ public SlotReport(final List<SlotStatus> slotsStatus) {
this.slotsStatus = checkNotNull(slotsStatus);
- this.resourceID = checkNotNull(resourceID);
}
public List<SlotStatus> getSlotsStatus() {
return slotsStatus;
}
- public ResourceID getResourceID() {
- return resourceID;
- }
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/0cac6614/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
----------------------------------------------------------------------
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 7df0a91..c0041a3 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
@@ -20,9 +20,12 @@ package org.apache.flink.runtime.taskexecutor;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
import org.apache.flink.runtime.clusterframework.types.ResourceID;
+import org.apache.flink.runtime.clusterframework.types.SlotID;
+import org.apache.flink.runtime.instance.InstanceID;
import org.apache.flink.runtime.metrics.MetricRegistry;
-import org.apache.flink.runtime.resourcemanager.SlotRequestRegistered;
-import org.apache.flink.runtime.resourcemanager.SlotRequestReply;
+import org.apache.flink.runtime.resourcemanager.messages.taskexecutor.TMSlotRequestRegistered;
+import org.apache.flink.runtime.resourcemanager.messages.taskexecutor.TMSlotRequestRejected;
+import org.apache.flink.runtime.resourcemanager.messages.taskexecutor.TMSlotRequestReply;
import org.apache.flink.runtime.rpc.FatalErrorHandler;
import org.apache.flink.runtime.taskmanager.TaskManagerLocation;
import org.apache.flink.annotation.VisibleForTesting;
@@ -38,6 +41,8 @@ import org.apache.flink.runtime.rpc.RpcService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.util.HashSet;
+import java.util.Set;
import java.util.UUID;
import static org.apache.flink.util.Preconditions.checkArgument;
@@ -78,6 +83,9 @@ public class TaskExecutor extends RpcEndpoint<TaskExecutorGateway> {
/** The fatal error handler to use in case of a fatal error */
private final FatalErrorHandler fatalErrorHandler;
+ /** Slots which have become available but haven't been confirmed by the RM */
+ private final Set<SlotID> unconfirmedFreeSlots;
+
// --------- resource manager --------
private TaskExecutorToResourceManagerConnection resourceManagerConnection;
@@ -109,6 +117,8 @@ public class TaskExecutor extends RpcEndpoint<TaskExecutorGateway> {
this.fatalErrorHandler = checkNotNull(fatalErrorHandler);
this.numberOfSlots = taskManagerConfiguration.getNumberSlots();
+
+ this.unconfirmedFreeSlots = new HashSet<>();
}
// ------------------------------------------------------------------------
@@ -152,6 +162,8 @@ public class TaskExecutor extends RpcEndpoint<TaskExecutorGateway> {
}
}
+ unconfirmedFreeSlots.clear();
+
// establish a connection to the new leader
if (newLeaderAddress != null) {
log.info("Attempting to register at ResourceManager {}", newLeaderAddress);
@@ -169,13 +181,25 @@ public class TaskExecutor extends RpcEndpoint<TaskExecutorGateway> {
/**
* Requests a slot from the TaskManager
*
+ * @param slotID Slot id for the request
* @param allocationID id for the request
* @param resourceManagerLeaderID current leader id of the ResourceManager
* @return answer to the slot request
*/
@RpcMethod
- public SlotRequestReply requestSlot(AllocationID allocationID, UUID resourceManagerLeaderID) {
- return new SlotRequestRegistered(allocationID);
+ public TMSlotRequestReply requestSlot(SlotID slotID, AllocationID allocationID, UUID resourceManagerLeaderID) {
+ if (!resourceManagerConnection.getTargetLeaderId().equals(resourceManagerLeaderID)) {
+ return new TMSlotRequestRejected(
+ resourceManagerConnection.getRegistrationId(), getResourceID(), allocationID);
+ }
+ if (unconfirmedFreeSlots.contains(slotID)) {
+ // check if request has not been blacklisted because the notification of a free slot
+ // has not been confirmed by the ResourceManager
+ return new TMSlotRequestRejected(
+ resourceManagerConnection.getRegistrationId(), getResourceID(), allocationID);
+ }
+ return new TMSlotRequestRegistered(new InstanceID(), ResourceID.generate(), allocationID);
+
}
// ------------------------------------------------------------------------
@@ -227,6 +251,11 @@ public class TaskExecutor extends RpcEndpoint<TaskExecutorGateway> {
return resourceManagerConnection;
}
+ @VisibleForTesting
+ public void addUnconfirmedFreeSlotNotification(SlotID slotID) {
+ unconfirmedFreeSlots.add(slotID);
+ }
+
// ------------------------------------------------------------------------
// Utility classes
// ------------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/0cac6614/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorGateway.java
----------------------------------------------------------------------
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 0962802..2360b53 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
@@ -20,8 +20,9 @@ package org.apache.flink.runtime.taskexecutor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.runtime.clusterframework.types.AllocationID;
+import org.apache.flink.runtime.clusterframework.types.SlotID;
import org.apache.flink.runtime.concurrent.Future;
-import org.apache.flink.runtime.resourcemanager.SlotRequestReply;
+import org.apache.flink.runtime.resourcemanager.messages.taskexecutor.TMSlotRequestReply;
import org.apache.flink.runtime.rpc.RpcGateway;
import org.apache.flink.runtime.rpc.RpcTimeout;
@@ -41,11 +42,13 @@ public interface TaskExecutorGateway extends RpcGateway {
/**
* Requests a slot from the TaskManager
*
+ * @param slotID slot id for the request
* @param allocationID id for the request
* @param resourceManagerLeaderID current leader id of the ResourceManager
* @return answer to the slot request
*/
- Future<SlotRequestReply> requestSlot(
+ Future<TMSlotRequestReply> requestSlot(
+ SlotID slotID,
AllocationID allocationID,
UUID resourceManagerLeaderID,
@RpcTimeout Time timeout);
http://git-wip-us.apache.org/repos/asf/flink/blob/0cac6614/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorToResourceManagerConnection.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorToResourceManagerConnection.java b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorToResourceManagerConnection.java
index b4b3bae..2dbd9eb 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorToResourceManagerConnection.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutorToResourceManagerConnection.java
@@ -113,7 +113,7 @@ public class TaskExecutorToResourceManagerConnection
ResourceManagerGateway resourceManager, UUID leaderId, long timeoutMillis) throws Exception {
Time timeout = Time.milliseconds(timeoutMillis);
- return resourceManager.registerTaskExecutor(leaderId, taskExecutorAddress, resourceID, timeout);
+ return resourceManager.registerTaskExecutor(leaderId, taskExecutorAddress, resourceID, new SlotReport(), timeout);
}
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/0cac6614/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerHATest.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerHATest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerHATest.java
index fdb83f5..ce1fdca 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerHATest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/resourcemanager/ResourceManagerHATest.java
@@ -21,6 +21,7 @@ package org.apache.flink.runtime.resourcemanager;
import org.apache.flink.runtime.highavailability.TestingHighAvailabilityServices;
import org.apache.flink.runtime.leaderelection.TestingLeaderElectionService;
import org.apache.flink.runtime.resourcemanager.slotmanager.SlotManager;
+import org.apache.flink.runtime.resourcemanager.slotmanager.SlotManagerFactory;
import org.apache.flink.runtime.rpc.MainThreadExecutable;
import org.apache.flink.runtime.rpc.RpcEndpoint;
import org.apache.flink.runtime.rpc.RpcGateway;
@@ -43,7 +44,8 @@ public class ResourceManagerHATest {
@Test
public void testGrantAndRevokeLeadership() throws Exception {
- // mock a RpcService which will return a special RpcGateway when call its startServer method, the returned RpcGateway directly execute runAsync call
+ // mock a RpcService which will return a special RpcGateway when call its startServer method,
+ // the returned RpcGateway directly executes runAsync calls
TestingResourceManagerGatewayProxy gateway = mock(TestingResourceManagerGatewayProxy.class);
doCallRealMethod().when(gateway).runAsync(any(Runnable.class));
@@ -54,18 +56,18 @@ public class ResourceManagerHATest {
TestingHighAvailabilityServices highAvailabilityServices = new TestingHighAvailabilityServices();
highAvailabilityServices.setResourceManagerLeaderElectionService(leaderElectionService);
- SlotManager slotManager = mock(SlotManager.class);
- final ResourceManager resourceManager = new StandaloneResourceManager(rpcService, highAvailabilityServices, slotManager);
+ final ResourceManager resourceManager =
+ new TestingResourceManager(rpcService, highAvailabilityServices);
resourceManager.start();
// before grant leadership, resourceManager's leaderId is null
- Assert.assertNull(resourceManager.getLeaderSessionID());
+ Assert.assertEquals(new UUID(0,0), resourceManager.getLeaderSessionID());
final UUID leaderId = UUID.randomUUID();
leaderElectionService.isLeader(leaderId);
// after grant leadership, resourceManager's leaderId has value
Assert.assertEquals(leaderId, resourceManager.getLeaderSessionID());
// then revoke leadership, resourceManager's leaderId is null again
leaderElectionService.notLeader();
- Assert.assertNull(resourceManager.getLeaderSessionID());
+ Assert.assertEquals(new UUID(0,0), resourceManager.getLeaderSessionID());
}
private static abstract class TestingResourceManagerGatewayProxy implements MainThreadExecutable, StartStoppable, RpcGateway {