protected void initializeServices(Configuration configuration) throws Exception {

		LOG.info("Initializing cluster services.");

		synchronized (lock) {
			final String bindAddress = configuration.getString(JobManagerOptions.ADDRESS);
			final String portRange = getRPCPortRange(configuration);

			commonRpcService = createRpcService(configuration, bindAddress, portRange);

			// update the configuration used to create the high availability services
			configuration.setString(JobManagerOptions.ADDRESS, commonRpcService.getAddress());
			configuration.setInteger(JobManagerOptions.PORT, commonRpcService.getPort());

			ioExecutor = Executors.newFixedThreadPool(
				Hardware.getNumberCPUCores(),
				new ExecutorThreadFactory("cluster-io"));
			haServices = createHaServices(configuration, ioExecutor);
			blobServer = new BlobServer(configuration, haServices.createBlobStore());
			blobServer.start();
			heartbeatServices = createHeartbeatServices(configuration);
			metricRegistry = createMetricRegistry(configuration);

			// TODO: This is a temporary hack until we have ported the MetricQueryService to the new RpcEndpoint
			// Start actor system for metric query service on any available port
			metricQueryServiceActorSystem = MetricUtils.startMetricsActorSystem(configuration, bindAddress, LOG);
			metricRegistry.startQueryService(metricQueryServiceActorSystem, null);

			archivedExecutionGraphStore = createSerializableExecutionGraphStore(configuration, commonRpcService.getScheduledExecutor());

			transientBlobCache = new TransientBlobCache(
				configuration,
				new InetSocketAddress(
					commonRpcService.getAddress(),
					blobServer.getPort()));
		}
	}
 

protected void initializeServices(Configuration configuration) throws Exception {

		LOG.info("Initializing cluster services.");

		synchronized (lock) {
			final String bindAddress = configuration.getString(JobManagerOptions.ADDRESS);
			final String portRange = getRPCPortRange(configuration);

			commonRpcService = createRpcService(configuration, bindAddress, portRange);

			// update the configuration used to create the high availability services
			configuration.setString(JobManagerOptions.ADDRESS, commonRpcService.getAddress());
			configuration.setInteger(JobManagerOptions.PORT, commonRpcService.getPort());

			ioExecutor = Executors.newFixedThreadPool(
				Hardware.getNumberCPUCores(),
				new ExecutorThreadFactory("cluster-io"));
			haServices = createHaServices(configuration, ioExecutor);
			blobServer = new BlobServer(configuration, haServices.createBlobStore());
			blobServer.start();
			heartbeatServices = createHeartbeatServices(configuration);
			metricRegistry = createMetricRegistry(configuration);

			final RpcService metricQueryServiceRpcService = MetricUtils.startMetricsRpcService(configuration, bindAddress);
			metricRegistry.startQueryService(metricQueryServiceRpcService, null);

			archivedExecutionGraphStore = createSerializableExecutionGraphStore(configuration, commonRpcService.getScheduledExecutor());
		}
	}
 

protected void initializeServices(Configuration configuration, PluginManager pluginManager) throws Exception {

		LOG.info("Initializing cluster services.");

		synchronized (lock) {
			commonRpcService = AkkaRpcServiceUtils.createRemoteRpcService(
				configuration,
				configuration.getString(JobManagerOptions.ADDRESS),
				getRPCPortRange(configuration),
				configuration.getString(JobManagerOptions.BIND_HOST),
				configuration.getOptional(JobManagerOptions.RPC_BIND_PORT));

			// update the configuration used to create the high availability services
			configuration.setString(JobManagerOptions.ADDRESS, commonRpcService.getAddress());
			configuration.setInteger(JobManagerOptions.PORT, commonRpcService.getPort());

			ioExecutor = Executors.newFixedThreadPool(
				ClusterEntrypointUtils.getPoolSize(configuration),
				new ExecutorThreadFactory("cluster-io"));
			haServices = createHaServices(configuration, ioExecutor);
			blobServer = new BlobServer(configuration, haServices.createBlobStore());
			blobServer.start();
			heartbeatServices = createHeartbeatServices(configuration);
			metricRegistry = createMetricRegistry(configuration, pluginManager);

			final RpcService metricQueryServiceRpcService = MetricUtils.startRemoteMetricsRpcService(configuration, commonRpcService.getAddress());
			metricRegistry.startQueryService(metricQueryServiceRpcService, null);

			final String hostname = RpcUtils.getHostname(commonRpcService);

			processMetricGroup = MetricUtils.instantiateProcessMetricGroup(
				metricRegistry,
				hostname,
				ConfigurationUtils.getSystemResourceMetricsProbingInterval(configuration));

			archivedExecutionGraphStore = createSerializableExecutionGraphStore(configuration, commonRpcService.getScheduledExecutor());
		}
	}
 

public TaskManagerRunner(Configuration configuration, ResourceID resourceId) throws Exception {
	this.configuration = checkNotNull(configuration);
	this.resourceId = checkNotNull(resourceId);

	timeout = AkkaUtils.getTimeoutAsTime(configuration);

	this.executor = java.util.concurrent.Executors.newScheduledThreadPool(
		Hardware.getNumberCPUCores(),
		new ExecutorThreadFactory("taskmanager-future"));

	highAvailabilityServices = HighAvailabilityServicesUtils.createHighAvailabilityServices(
		configuration,
		executor,
		HighAvailabilityServicesUtils.AddressResolution.TRY_ADDRESS_RESOLUTION);

	rpcService = createRpcService(configuration, highAvailabilityServices);
	metricQueryServiceActorSystem = MetricUtils.startMetricsActorSystem(configuration, rpcService.getAddress(), LOG);

	HeartbeatServices heartbeatServices = HeartbeatServices.fromConfiguration(configuration);

	metricRegistry = new MetricRegistryImpl(MetricRegistryConfiguration.fromConfiguration(configuration));

	// TODO: Temporary hack until the MetricQueryService has been ported to RpcEndpoint
	metricRegistry.startQueryService(metricQueryServiceActorSystem, resourceId);

	blobCacheService = new BlobCacheService(
		configuration, highAvailabilityServices.createBlobStore(), null
	);

	taskManager = startTaskManager(
		this.configuration,
		this.resourceId,
		rpcService,
		highAvailabilityServices,
		heartbeatServices,
		metricRegistry,
		blobCacheService,
		false,
		this);

	this.terminationFuture = new CompletableFuture<>();
	this.shutdown = false;

	MemoryLogger.startIfConfigured(LOG, configuration, metricQueryServiceActorSystem);
}
 

public static TaskExecutor startTaskManager(
		Configuration configuration,
		ResourceID resourceID,
		RpcService rpcService,
		HighAvailabilityServices highAvailabilityServices,
		HeartbeatServices heartbeatServices,
		MetricRegistry metricRegistry,
		BlobCacheService blobCacheService,
		boolean localCommunicationOnly,
		FatalErrorHandler fatalErrorHandler) throws Exception {

	checkNotNull(configuration);
	checkNotNull(resourceID);
	checkNotNull(rpcService);
	checkNotNull(highAvailabilityServices);

	LOG.info("Starting TaskManager with ResourceID: {}", resourceID);

	InetAddress remoteAddress = InetAddress.getByName(rpcService.getAddress());

	TaskManagerServicesConfiguration taskManagerServicesConfiguration =
		TaskManagerServicesConfiguration.fromConfiguration(
			configuration,
			remoteAddress,
			localCommunicationOnly);

	TaskManagerServices taskManagerServices = TaskManagerServices.fromConfiguration(
		taskManagerServicesConfiguration,
		resourceID,
		rpcService.getExecutor(), // TODO replace this later with some dedicated executor for io.
		EnvironmentInformation.getSizeOfFreeHeapMemoryWithDefrag(),
		EnvironmentInformation.getMaxJvmHeapMemory());

	TaskManagerMetricGroup taskManagerMetricGroup = MetricUtils.instantiateTaskManagerMetricGroup(
		metricRegistry,
		taskManagerServices.getTaskManagerLocation(),
		taskManagerServices.getNetworkEnvironment(),
		taskManagerServicesConfiguration.getSystemResourceMetricsProbingInterval());

	TaskManagerConfiguration taskManagerConfiguration = TaskManagerConfiguration.fromConfiguration(configuration);

	String metricQueryServicePath = metricRegistry.getMetricQueryServicePath();

	return new TaskExecutor(
		rpcService,
		taskManagerConfiguration,
		highAvailabilityServices,
		taskManagerServices,
		heartbeatServices,
		taskManagerMetricGroup,
		metricQueryServicePath,
		blobCacheService,
		fatalErrorHandler);
}
 

public TaskManagerRunner(Configuration configuration, ResourceID resourceId) throws Exception {
	this.configuration = checkNotNull(configuration);
	this.resourceId = checkNotNull(resourceId);

	timeout = AkkaUtils.getTimeoutAsTime(configuration);

	this.executor = java.util.concurrent.Executors.newScheduledThreadPool(
		Hardware.getNumberCPUCores(),
		new ExecutorThreadFactory("taskmanager-future"));

	highAvailabilityServices = HighAvailabilityServicesUtils.createHighAvailabilityServices(
		configuration,
		executor,
		HighAvailabilityServicesUtils.AddressResolution.TRY_ADDRESS_RESOLUTION);

	rpcService = createRpcService(configuration, highAvailabilityServices);

	HeartbeatServices heartbeatServices = HeartbeatServices.fromConfiguration(configuration);

	metricRegistry = new MetricRegistryImpl(
		MetricRegistryConfiguration.fromConfiguration(configuration),
		ReporterSetup.fromConfiguration(configuration));

	final RpcService metricQueryServiceRpcService = MetricUtils.startMetricsRpcService(configuration, rpcService.getAddress());
	metricRegistry.startQueryService(metricQueryServiceRpcService, resourceId);

	blobCacheService = new BlobCacheService(
		configuration, highAvailabilityServices.createBlobStore(), null
	);

	taskManager = startTaskManager(
		this.configuration,
		this.resourceId,
		rpcService,
		highAvailabilityServices,
		heartbeatServices,
		metricRegistry,
		blobCacheService,
		false,
		this);

	this.terminationFuture = new CompletableFuture<>();
	this.shutdown = false;

	MemoryLogger.startIfConfigured(LOG, configuration, terminationFuture);
}
 

public static TaskExecutor startTaskManager(
		Configuration configuration,
		ResourceID resourceID,
		RpcService rpcService,
		HighAvailabilityServices highAvailabilityServices,
		HeartbeatServices heartbeatServices,
		MetricRegistry metricRegistry,
		BlobCacheService blobCacheService,
		boolean localCommunicationOnly,
		FatalErrorHandler fatalErrorHandler) throws Exception {

	checkNotNull(configuration);
	checkNotNull(resourceID);
	checkNotNull(rpcService);
	checkNotNull(highAvailabilityServices);

	LOG.info("Starting TaskManager with ResourceID: {}", resourceID);

	InetAddress remoteAddress = InetAddress.getByName(rpcService.getAddress());

	TaskManagerServicesConfiguration taskManagerServicesConfiguration =
		TaskManagerServicesConfiguration.fromConfiguration(
			configuration,
			resourceID,
			remoteAddress,
			EnvironmentInformation.getSizeOfFreeHeapMemoryWithDefrag(),
			EnvironmentInformation.getMaxJvmHeapMemory(),
			localCommunicationOnly);

	Tuple2<TaskManagerMetricGroup, MetricGroup> taskManagerMetricGroup = MetricUtils.instantiateTaskManagerMetricGroup(
		metricRegistry,
		TaskManagerLocation.getHostName(remoteAddress),
		resourceID,
		taskManagerServicesConfiguration.getSystemResourceMetricsProbingInterval());

	TaskManagerServices taskManagerServices = TaskManagerServices.fromConfiguration(
		taskManagerServicesConfiguration,
		taskManagerMetricGroup.f1,
		rpcService.getExecutor()); // TODO replace this later with some dedicated executor for io.

	TaskManagerConfiguration taskManagerConfiguration = TaskManagerConfiguration.fromConfiguration(configuration);

	String metricQueryServiceAddress = metricRegistry.getMetricQueryServiceGatewayRpcAddress();

	return new TaskExecutor(
		rpcService,
		taskManagerConfiguration,
		highAvailabilityServices,
		taskManagerServices,
		heartbeatServices,
		taskManagerMetricGroup.f0,
		metricQueryServiceAddress,
		blobCacheService,
		fatalErrorHandler,
		new PartitionTable<>());
}
 

public TaskManagerRunner(Configuration configuration, PluginManager pluginManager) throws Exception {
	this.configuration = checkNotNull(configuration);

	timeout = AkkaUtils.getTimeoutAsTime(configuration);

	this.executor = java.util.concurrent.Executors.newScheduledThreadPool(
		Hardware.getNumberCPUCores(),
		new ExecutorThreadFactory("taskmanager-future"));

	highAvailabilityServices = HighAvailabilityServicesUtils.createHighAvailabilityServices(
		configuration,
		executor,
		HighAvailabilityServicesUtils.AddressResolution.NO_ADDRESS_RESOLUTION);

	rpcService = createRpcService(configuration, highAvailabilityServices);

	this.resourceId = new ResourceID(getTaskManagerResourceID(configuration, rpcService.getAddress(), rpcService.getPort()));

	HeartbeatServices heartbeatServices = HeartbeatServices.fromConfiguration(configuration);

	metricRegistry = new MetricRegistryImpl(
		MetricRegistryConfiguration.fromConfiguration(configuration),
		ReporterSetup.fromConfiguration(configuration, pluginManager));

	final RpcService metricQueryServiceRpcService = MetricUtils.startRemoteMetricsRpcService(configuration, rpcService.getAddress());
	metricRegistry.startQueryService(metricQueryServiceRpcService, resourceId);

	blobCacheService = new BlobCacheService(
		configuration, highAvailabilityServices.createBlobStore(), null
	);

	final ExternalResourceInfoProvider externalResourceInfoProvider =
		ExternalResourceUtils.createStaticExternalResourceInfoProvider(
			ExternalResourceUtils.getExternalResourceAmountMap(configuration),
			ExternalResourceUtils.externalResourceDriversFromConfig(configuration, pluginManager));

	taskManager = startTaskManager(
		this.configuration,
		this.resourceId,
		rpcService,
		highAvailabilityServices,
		heartbeatServices,
		metricRegistry,
		blobCacheService,
		false,
		externalResourceInfoProvider,
		this);

	this.terminationFuture = new CompletableFuture<>();
	this.shutdown = false;

	MemoryLogger.startIfConfigured(LOG, configuration, terminationFuture);
}
 

public static TaskExecutor startTaskManager(
		Configuration configuration,
		ResourceID resourceID,
		RpcService rpcService,
		HighAvailabilityServices highAvailabilityServices,
		HeartbeatServices heartbeatServices,
		MetricRegistry metricRegistry,
		BlobCacheService blobCacheService,
		boolean localCommunicationOnly,
		ExternalResourceInfoProvider externalResourceInfoProvider,
		FatalErrorHandler fatalErrorHandler) throws Exception {

	checkNotNull(configuration);
	checkNotNull(resourceID);
	checkNotNull(rpcService);
	checkNotNull(highAvailabilityServices);

	LOG.info("Starting TaskManager with ResourceID: {}", resourceID);

	String externalAddress = rpcService.getAddress();

	final TaskExecutorResourceSpec taskExecutorResourceSpec = TaskExecutorResourceUtils.resourceSpecFromConfig(configuration);

	TaskManagerServicesConfiguration taskManagerServicesConfiguration =
		TaskManagerServicesConfiguration.fromConfiguration(
			configuration,
			resourceID,
			externalAddress,
			localCommunicationOnly,
			taskExecutorResourceSpec);

	Tuple2<TaskManagerMetricGroup, MetricGroup> taskManagerMetricGroup = MetricUtils.instantiateTaskManagerMetricGroup(
		metricRegistry,
		externalAddress,
		resourceID,
		taskManagerServicesConfiguration.getSystemResourceMetricsProbingInterval());

	final ExecutorService ioExecutor = Executors.newFixedThreadPool(
		taskManagerServicesConfiguration.getNumIoThreads(),
		new ExecutorThreadFactory("flink-taskexecutor-io"));

	TaskManagerServices taskManagerServices = TaskManagerServices.fromConfiguration(
		taskManagerServicesConfiguration,
		blobCacheService.getPermanentBlobService(),
		taskManagerMetricGroup.f1,
		ioExecutor,
		fatalErrorHandler);

	TaskManagerConfiguration taskManagerConfiguration =
		TaskManagerConfiguration.fromConfiguration(configuration, taskExecutorResourceSpec, externalAddress);

	String metricQueryServiceAddress = metricRegistry.getMetricQueryServiceGatewayRpcAddress();

	return new TaskExecutor(
		rpcService,
		taskManagerConfiguration,
		highAvailabilityServices,
		taskManagerServices,
		externalResourceInfoProvider,
		heartbeatServices,
		taskManagerMetricGroup.f0,
		metricQueryServiceAddress,
		blobCacheService,
		fatalErrorHandler,
		new TaskExecutorPartitionTrackerImpl(taskManagerServices.getShuffleEnvironment()),
		createBackPressureSampleService(configuration, rpcService.getScheduledExecutor()));
}