MesosFlinkResourceManager extends FlinkResourceManager<RegisteredMesosWorkerNode> { @Override protected void initialize() throws Exception { LOG.info("Initializing Mesos resource master"); workerStore.start(); schedulerCallbackHandler = new SchedulerProxy(self()); FrameworkInfo.Builder frameworkInfo = mesosConfig.frameworkInfo() .clone() .setCheckpoint(true); 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()); } MesosConfiguration initializedMesosConfig = mesosConfig.withFrameworkInfo(frameworkInfo); MesosConfiguration.logMesosConfig(LOG, initializedMesosConfig); schedulerDriver = initializedMesosConfig.createDriver(schedulerCallbackHandler, false); connectionMonitor = createConnectionMonitor(); launchCoordinator = createLaunchCoordinator(); reconciliationCoordinator = createReconciliationCoordinator(); taskRouter = createTaskRouter(); recoverWorkers(); connectionMonitor.tell(new ConnectionMonitor.Start(), self()); schedulerDriver.start(); } MesosFlinkResourceManager(
		Configuration flinkConfig,
		MesosConfiguration mesosConfig,
		MesosWorkerStore workerStore,
		LeaderRetrievalService leaderRetrievalService,
		MesosTaskManagerParameters taskManagerParameters,
		ContainerSpecification taskManagerContainerSpec,
		MesosArtifactResolver artifactResolver,
		int maxFailedTasks,
		int numInitialTaskManagers); @Override void postStop(); static Props createActorProps(
			Class<? extends MesosFlinkResourceManager> actorClass,
			Configuration flinkConfig,
			MesosConfiguration mesosConfig,
			MesosWorkerStore workerStore,
			LeaderRetrievalService leaderRetrievalService,
			MesosTaskManagerParameters taskManagerParameters,
			ContainerSpecification taskManagerContainerSpec,
			MesosArtifactResolver artifactResolver,
			Logger log);  }
@Test public void testOfferHandling() { new Context() {{ new Within(duration("10 seconds")) { @Override protected void run() { initialize(); register(Collections.<ResourceID>emptyList()); resourceManager.tell(new ResourceOffers(Collections.<Protos.Offer>emptyList())); resourceManagerInstance.launchCoordinator.expectMsgClass(ResourceOffers.class); resourceManager.tell(new OfferRescinded(offer1)); resourceManagerInstance.launchCoordinator.expectMsgClass(OfferRescinded.class); } }; }}; }
@Test public void testStatusHandling() { new Context() {{ new Within(duration("10 seconds")) { @Override protected void run() { initialize(); register(Collections.<ResourceID>emptyList()); resourceManager.tell(new StatusUpdate(Protos.TaskStatus.newBuilder() .setTaskId(task1).setSlaveId(slave1).setState(Protos.TaskState.TASK_LOST).build()), resourceManager); resourceManagerInstance.reconciliationCoordinator.expectMsgClass(StatusUpdate.class); resourceManagerInstance.taskRouter.expectMsgClass(StatusUpdate.class); } }; }}; }
@Test public void testStopApplication() { new Context() {{ new Within(duration("10 seconds")) { @Override protected void run() { try { initialize(); register(Collections.<ResourceID>emptyList()); watch(resourceManager.actor()); resourceManager.tell(new StopCluster(ApplicationStatus.SUCCEEDED, ""), resourceManager); verify(schedulerDriver).stop(false); verify(workerStore).stop(true); expectTerminated(resourceManager.actor()); } catch (Exception ex) { throw new RuntimeException(ex); } } }; }}; }