MesosResourceManager extends ResourceManager<RegisteredMesosWorkerNode> { @Override protected void initialize() throws ResourceManagerException { try { this.workerStore = mesosServices.createMesosWorkerStore(flinkConfig, getRpcService().getExecutor()); workerStore.start(); } catch (Exception e) { throw new ResourceManagerException("Unable to initialize the worker store.", e); } Protos.FrameworkInfo.Builder frameworkInfo = mesosConfig.frameworkInfo() .clone() .setCheckpoint(true); try { Option<Protos.FrameworkID> frameworkID = workerStore.getFrameworkID(); if (frameworkID.isEmpty()) { LOG.info("Registering as new framework."); } else { LOG.info("Recovery scenario: re-registering using framework ID {}.", frameworkID.get().getValue()); frameworkInfo.setId(frameworkID.get()); } } catch (Exception e) { throw new ResourceManagerException("Unable to recover the framework ID.", e); } MesosConfiguration initializedMesosConfig = mesosConfig.withFrameworkInfo(frameworkInfo); MesosConfiguration.logMesosConfig(LOG, initializedMesosConfig); schedulerDriver = initializedMesosConfig.createDriver( new MesosResourceManagerSchedulerCallback(), false); selfActor = createSelfActor(); connectionMonitor = createConnectionMonitor(); launchCoordinator = createLaunchCoordinator(schedulerDriver, selfActor); reconciliationCoordinator = createReconciliationCoordinator(schedulerDriver); taskMonitor = createTaskMonitor(schedulerDriver); try { recoverWorkers(); } catch (Exception e) { throw new ResourceManagerException("Unable to recover Mesos worker state.", e); } try { LaunchableMesosWorker.configureArtifactServer(artifactServer, taskManagerContainerSpec); } catch (IOException e) { throw new ResourceManagerException("Unable to configure the artifact server with TaskManager artifacts.", e); } connectionMonitor.tell(new ConnectionMonitor.Start(), selfActor); schedulerDriver.start(); LOG.info("Mesos resource manager initialized."); } 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 testInitialize() throws Exception { new Context() {{ startResourceManager(); LOG.info("initialized"); }}; }