@Test public void testInitialize() throws Exception { new Context() {{ startResourceManager(); LOG.info("initialized"); }}; }
@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); if (webUiUrl != null) { frameworkInfo.setWebuiUrl(webUiUrl); } 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); } initializedMesosConfig = mesosConfig.withFrameworkInfo(frameworkInfo); MesosConfiguration.logMesosConfig(LOG, initializedMesosConfig); this.selfActor = createSelfActor(); try { LaunchableMesosWorker.configureArtifactServer(artifactServer, taskManagerContainerSpec); } catch (IOException e) { throw new ResourceManagerException("Unable to configure the artifact server with TaskManager artifacts.", e); } }
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); if (webUiUrl != null) { frameworkInfo.setWebuiUrl(webUiUrl); } 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); } initializedMesosConfig = mesosConfig.withFrameworkInfo(frameworkInfo); MesosConfiguration.logMesosConfig(LOG, initializedMesosConfig); this.selfActor = createSelfActor(); try { LaunchableMesosWorker.configureArtifactServer(artifactServer, taskManagerContainerSpec); } catch (IOException e) { throw new ResourceManagerException("Unable to configure the artifact server with TaskManager artifacts.", e); } } }
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); if (webUiUrl != null) { frameworkInfo.setWebuiUrl(webUiUrl); } 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); } initializedMesosConfig = mesosConfig.withFrameworkInfo(frameworkInfo); MesosConfiguration.logMesosConfig(LOG, initializedMesosConfig); this.selfActor = createSelfActor(); try { LaunchableMesosWorker.configureArtifactServer(artifactServer, taskManagerContainerSpec); } catch (IOException e) { throw new ResourceManagerException("Unable to configure the artifact server with TaskManager artifacts.", e); } } MesosResourceManager(
			// base class
			RpcService rpcService,
			String resourceManagerEndpointId,
			ResourceID resourceId,
			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,
			@Nullable String webUiUrl,
			JobManagerMetricGroup jobManagerMetricGroup); }
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); if (webUiUrl != null) { frameworkInfo.setWebuiUrl(webUiUrl); } 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); } initializedMesosConfig = mesosConfig.withFrameworkInfo(frameworkInfo); MesosConfiguration.logMesosConfig(LOG, initializedMesosConfig); this.selfActor = createSelfActor(); try { LaunchableMesosWorker.configureArtifactServer(artifactServer, taskManagerContainerSpec); } catch (IOException e) { throw new ResourceManagerException("Unable to configure the artifact server with TaskManager artifacts.", e); } } MesosResourceManager(
			// base class
			RpcService rpcService,
			String resourceManagerEndpointId,
			ResourceID resourceId,
			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,
			@Nullable String webUiUrl,
			JobManagerMetricGroup jobManagerMetricGroup); @Override CompletableFuture<Void> onStop(); @Override Collection<ResourceProfile> startNewWorker(ResourceProfile resourceProfile); @Override boolean stopWorker(RegisteredMesosWorkerNode workerNode); void acceptOffers(AcceptOffers msg); void reconcile(ReconciliationCoordinator.Reconcile message); void taskTerminated(TaskMonitor.TaskTerminated message); }
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); if (webUiUrl != null) { frameworkInfo.setWebuiUrl(webUiUrl); } 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); } initializedMesosConfig = mesosConfig.withFrameworkInfo(frameworkInfo); MesosConfiguration.logMesosConfig(LOG, initializedMesosConfig); this.selfActor = createSelfActor(); try { LaunchableMesosWorker.configureArtifactServer(artifactServer, taskManagerContainerSpec); } catch (IOException e) { throw new ResourceManagerException("Unable to configure the artifact server with TaskManager artifacts.", e); } } MesosResourceManager(
			// base class
			RpcService rpcService,
			String resourceManagerEndpointId,
			ResourceID resourceId,
			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,
			@Nullable String webUiUrl,
			JobManagerMetricGroup jobManagerMetricGroup); @Override CompletableFuture<Void> onStop(); @Override Collection<ResourceProfile> startNewWorker(ResourceProfile resourceProfile); @Override boolean stopWorker(RegisteredMesosWorkerNode workerNode); void acceptOffers(AcceptOffers msg); void reconcile(ReconciliationCoordinator.Reconcile message); void taskTerminated(TaskMonitor.TaskTerminated message);  }