MesosResourceManager extends ResourceManager<RegisteredMesosWorkerNode> { @Override protected void shutDownApplication( ApplicationStatus finalStatus, String optionalDiagnostics) throws ResourceManagerException { LOG.info("Shutting down and unregistering as a Mesos framework."); Exception exception = null; try { schedulerDriver.stop(false); } catch (Exception ex) { exception = new Exception("Could not unregister the Mesos framework.", ex); } try { workerStore.stop(true); } catch (Exception ex) { exception = ExceptionUtils.firstOrSuppressed( new Exception("Could not stop the Mesos worker store.", ex), exception); } if (exception != null) { throw new ResourceManagerException("Could not properly shut down the Mesos application.", exception); } } MesosResourceManager(
			// base class
			RpcService rpcService,
			String resourceManagerEndpointId,
			ResourceID resourceId,
			ResourceManagerConfiguration resourceManagerConfiguration,
			HighAvailabilityServices highAvailabilityServices,
			HeartbeatServices heartbeatServices,
			SlotManager slotManager,
			MetricRegistry metricRegistry,
			JobLeaderIdService jobLeaderIdService,
			FatalErrorHandler fatalErrorHandler,
			// Mesos specifics
			Configuration flinkConfig,
			MesosServices mesosServices,
			MesosConfiguration mesosConfig,
			MesosTaskManagerParameters taskManagerParameters,
			ContainerSpecification taskManagerContainerSpec); @Override void postStop(); @Override void startNewWorker(ResourceProfile resourceProfile); @Override boolean stopWorker(ResourceID resourceID); void acceptOffers(AcceptOffers msg); void reconcile(ReconciliationCoordinator.Reconcile message); void taskTerminated(TaskMonitor.TaskTerminated message);  }
@Test public void testShutdownApplication() throws Exception { new Context() {{ startResourceManager(); resourceManager.shutDownCluster(ApplicationStatus.SUCCEEDED, ""); verify(rmServices.schedulerDriver).stop(false); verify(rmServices.workerStore).stop(true); }}; }