@Test public void testStopWorker() throws Exception { new Context() {{ MesosWorkerStore.Worker worker1launched = MesosWorkerStore.Worker.newWorker(task1).launchWorker(slave1, slave1host); when(rmServices.workerStore.getFrameworkID()).thenReturn(Option.apply(framework1)); when(rmServices.workerStore.recoverWorkers()).thenReturn(singletonList(worker1launched)); startResourceManager(); resourceManager.launchCoordinator.expectMsgClass(LaunchCoordinator.Assign.class); resourceManager.stopWorker(new RegisteredMesosWorkerNode(worker1launched)); MesosWorkerStore.Worker worker1Released = worker1launched.releaseWorker(); verify(rmServices.workerStore).putWorker(worker1Released); assertThat(resourceManager.workersInLaunch.entrySet(), empty()); assertThat(resourceManager.workersBeingReturned, hasEntry(extractResourceID(task1), worker1Released)); resourceManager.taskRouter.expectMsgClass(TaskMonitor.TaskGoalStateUpdated.class); resourceManager.launchCoordinator.expectMsgClass(LaunchCoordinator.Unassign.class); }}; }
@Override public boolean stopWorker(RegisteredMesosWorkerNode workerNode) { LOG.info("Stopping worker {}.", workerNode.getResourceID()); try { if (workersInLaunch.containsKey(workerNode.getResourceID())) { 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()) { 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; }
MesosResourceManager extends ResourceManager<RegisteredMesosWorkerNode> { @Override public boolean stopWorker(RegisteredMesosWorkerNode workerNode) { LOG.info("Stopping worker {}.", workerNode.getResourceID()); try { if (workersInLaunch.containsKey(workerNode.getResourceID())) { 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()) { 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; } }
MesosResourceManager extends ResourceManager<RegisteredMesosWorkerNode> { @Override public boolean stopWorker(RegisteredMesosWorkerNode workerNode) { LOG.info("Stopping worker {}.", workerNode.getResourceID()); try { if (workersInLaunch.containsKey(workerNode.getResourceID())) { 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()) { 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; } 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); }
MesosResourceManager extends ResourceManager<RegisteredMesosWorkerNode> { @Override public boolean stopWorker(RegisteredMesosWorkerNode workerNode) { LOG.info("Stopping worker {}.", workerNode.getResourceID()); try { if (workersInLaunch.containsKey(workerNode.getResourceID())) { 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()) { 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; } 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); @Override CompletableFuture<Void> onStop(); @Override Collection<ResourceProfile> startNewWorker(ResourceProfile resourceProfile); @Override boolean stopWorker(RegisteredMesosWorkerNode workerNode); void acceptOffers(AcceptOffers msg); void reconcile(ReconciliationCoordinator.Reconcile message); void taskTerminated(TaskMonitor.TaskTerminated message); }
MesosResourceManager extends ResourceManager<RegisteredMesosWorkerNode> { @Override public boolean stopWorker(RegisteredMesosWorkerNode workerNode) { LOG.info("Stopping worker {}.", workerNode.getResourceID()); try { if (workersInLaunch.containsKey(workerNode.getResourceID())) { 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()) { 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; } 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); @Override CompletableFuture<Void> onStop(); @Override Collection<ResourceProfile> startNewWorker(ResourceProfile resourceProfile); @Override boolean stopWorker(RegisteredMesosWorkerNode workerNode); void acceptOffers(AcceptOffers msg); void reconcile(ReconciliationCoordinator.Reconcile message); void taskTerminated(TaskMonitor.TaskTerminated message);  }