class MesosResourceManager extends ResourceManager<RegisteredMesosWorkerNode> {

	@Override
	public boolean stopWorker(RegisteredMesosWorkerNode workerNode) {
		LOG.info("Stopping worker {}.", workerNode.getResourceID());
		try {

			if (workersInLaunch.containsKey(workerNode.getResourceID())) {
				// update persistent state of worker to Released
				MesosWorkerStore.Worker worker = workersInLaunch.remove(workerNode.getResourceID());
				worker = worker.releaseWorker();
				workerStore.putWorker(worker);
				workersBeingReturned.put(extractResourceID(worker.taskID()), worker);

				taskMonitor.tell(new TaskMonitor.TaskGoalStateUpdated(extractGoalState(worker)), selfActor);

				if (worker.hostname().isDefined()) {
					// tell the launch coordinator that the task is being unassigned from the host, for planning purposes
					launchCoordinator.tell(new LaunchCoordinator.Unassign(worker.taskID(), worker.hostname().get()), selfActor);
				}
			}
			else if (workersBeingReturned.containsKey(workerNode.getResourceID())) {
				LOG.info("Ignoring request to stop worker {} because it is already being stopped.", workerNode.getResourceID());
			}
			else {
				LOG.warn("Unrecognized worker {}.", workerNode.getResourceID());
			}
		}
		catch (Exception e) {
			onFatalError(new ResourceManagerException("Unable to release a worker.", e));
		}

		return true;
	}

	public  MesosResourceManager(
			// base class
			RpcService rpcService,
			String resourceManagerEndpointId,
			ResourceID resourceId,
			HighAvailabilityServices highAvailabilityServices,
			HeartbeatServices heartbeatServices,
			SlotManager slotManager,
			MetricRegistry metricRegistry,
			JobLeaderIdService jobLeaderIdService,
			ClusterInformation clusterInformation,
			FatalErrorHandler fatalErrorHandler,
			// Mesos specifics
			Configuration flinkConfig,
			MesosServices mesosServices,
			MesosConfiguration mesosConfig,
			MesosTaskManagerParameters taskManagerParameters,
			ContainerSpecification taskManagerContainerSpec,
			@Nullable String webUiUrl,
			JobManagerMetricGroup jobManagerMetricGroup);

}