class MesosFlinkResourceManager extends FlinkResourceManager<RegisteredMesosWorkerNode> {

	@Override
	protected void initialize() throws Exception {
		LOG.info("Initializing Mesos resource master");

		workerStore.start();

		// create the scheduler driver to communicate with Mesos
		schedulerCallbackHandler = new SchedulerProxy(self());

		// register with Mesos
		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());
		}

		if (taskManagerParameters.gpus() > 0) {
			LOG.info("Add GPU_RESOURCES capability to framework");
			frameworkInfo.addCapabilities(Capability.newBuilder().setType(Capability.Type.GPU_RESOURCES));
		}

		MesosConfiguration initializedMesosConfig = mesosConfig.withFrameworkInfo(frameworkInfo);
		MesosConfiguration.logMesosConfig(LOG, initializedMesosConfig);
		schedulerDriver = initializedMesosConfig.createDriver(schedulerCallbackHandler, false);

		// create supporting actors
		connectionMonitor = createConnectionMonitor();
		launchCoordinator = createLaunchCoordinator();
		reconciliationCoordinator = createReconciliationCoordinator();
		taskRouter = createTaskRouter();

		recoverWorkers();

		connectionMonitor.tell(new ConnectionMonitor.Start(), self());
		schedulerDriver.start();
	}

	public  MesosFlinkResourceManager(
		Configuration flinkConfig,
		MesosConfiguration mesosConfig,
		MesosWorkerStore workerStore,
		LeaderRetrievalService leaderRetrievalService,
		MesosTaskManagerParameters taskManagerParameters,
		ContainerSpecification taskManagerContainerSpec,
		MesosArtifactResolver artifactResolver,
		int maxFailedTasks,
		int numInitialTaskManagers);

}