class TaskManagerServicesConfiguration {

	public static TaskManagerServicesConfiguration fromConfiguration(
			Configuration configuration,
			ResourceID resourceID,
			String externalAddress,
			boolean localCommunicationOnly,
			TaskExecutorResourceSpec taskExecutorResourceSpec) throws Exception {
		final String[] tmpDirs = ConfigurationUtils.parseTempDirectories(configuration);
		String[] localStateRootDir = ConfigurationUtils.parseLocalStateDirectories(configuration);
		if (localStateRootDir.length == 0) {
			// default to temp dirs.
			localStateRootDir = tmpDirs;
		}

		boolean localRecoveryMode = configuration.getBoolean(CheckpointingOptions.LOCAL_RECOVERY);

		final QueryableStateConfiguration queryableStateConfig = QueryableStateConfiguration.fromConfiguration(configuration);

		long timerServiceShutdownTimeout = AkkaUtils.getTimeout(configuration).toMillis();

		final RetryingRegistrationConfiguration retryingRegistrationConfiguration = RetryingRegistrationConfiguration.fromConfiguration(configuration);

		final int externalDataPort = configuration.getInteger(NettyShuffleEnvironmentOptions.DATA_PORT);

		String bindAddr = configuration.getString(TaskManagerOptions.BIND_HOST, NetUtils.getWildcardIPAddress());
		InetAddress bindAddress = InetAddress.getByName(bindAddr);

		final String classLoaderResolveOrder =
			configuration.getString(CoreOptions.CLASSLOADER_RESOLVE_ORDER);

		final String[] alwaysParentFirstLoaderPatterns = CoreOptions.getParentFirstLoaderPatterns(configuration);

		final int numIoThreads = ClusterEntrypointUtils.getPoolSize(configuration);

		return new TaskManagerServicesConfiguration(
			configuration,
			resourceID,
			externalAddress,
			bindAddress,
			externalDataPort,
			localCommunicationOnly,
			tmpDirs,
			localStateRootDir,
			localRecoveryMode,
			queryableStateConfig,
			ConfigurationParserUtils.getSlot(configuration),
			ConfigurationParserUtils.getPageSize(configuration),
			taskExecutorResourceSpec,
			timerServiceShutdownTimeout,
			retryingRegistrationConfiguration,
			ConfigurationUtils.getSystemResourceMetricsProbingInterval(configuration),
			FlinkUserCodeClassLoaders.ResolveOrder.fromString(classLoaderResolveOrder),
			alwaysParentFirstLoaderPatterns,
			numIoThreads);
	}

	private  TaskManagerServicesConfiguration(
			Configuration configuration,
			ResourceID resourceID,
			String externalAddress,
			InetAddress bindAddress,
			int externalDataPort,
			boolean localCommunicationOnly,
			String[] tmpDirPaths,
			String[] localRecoveryStateRootDirectories,
			boolean localRecoveryEnabled,
			@Nullable QueryableStateConfiguration queryableStateConfig,
			int numberOfSlots,
			int pageSize,
			TaskExecutorResourceSpec taskExecutorResourceSpec,
			long timerServiceShutdownTimeout,
			RetryingRegistrationConfiguration retryingRegistrationConfiguration,
			Optional<Time> systemResourceMetricsProbingInterval,
			FlinkUserCodeClassLoaders.ResolveOrder classLoaderResolveOrder,
			String[] alwaysParentFirstLoaderPatterns,
			int numIoThreads);

}