class MesosResourceManager extends ResourceManager<RegisteredMesosWorkerNode> {

	@Override
	protected void initialize() throws ResourceManagerException {
		// create and start the worker store
		try {
			this.workerStore = mesosServices.createMesosWorkerStore(flinkConfig, getRpcService().getExecutor());
			workerStore.start();
		} catch (Exception e) {
			throw new ResourceManagerException("Unable to initialize the worker store.", e);
		}

		// register with Mesos
		// TODO : defer connection until RM acquires leadership

		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);

		// create supporting actors
		selfActor = createSelfActor();
		connectionMonitor = createConnectionMonitor();
		launchCoordinator = createLaunchCoordinator(schedulerDriver, selfActor);
		reconciliationCoordinator = createReconciliationCoordinator(schedulerDriver);
		taskMonitor = createTaskMonitor(schedulerDriver);

		// recover state
		try {
			recoverWorkers();
		} catch (Exception e) {
			throw new ResourceManagerException("Unable to recover Mesos worker state.", e);
		}

		// configure the artifact server to serve the TM container artifacts
		try {
			LaunchableMesosWorker.configureArtifactServer(artifactServer, taskManagerContainerSpec);
		}
		catch (IOException e) {
			throw new ResourceManagerException("Unable to configure the artifact server with TaskManager artifacts.", e);
		}

		// begin scheduling
		connectionMonitor.tell(new ConnectionMonitor.Start(), selfActor);
		schedulerDriver.start();

		LOG.info("Mesos resource manager initialized.");
	}

	public  MesosResourceManager(
			// base class
			RpcService rpcService,
			String resourceManagerEndpointId,
			ResourceID resourceId,
			ResourceManagerConfiguration resourceManagerConfiguration,
			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);

}