code: 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,
  				EnvironmentInformation.getMaxJvmHeapMemory(),
  				remoteAddress,
  				localCommunicationOnly);
  
  		TaskManagerMetricGroup taskManagerMetricGroup = MetricUtils.instantiateTaskManagerMetricGroup(
  			metricRegistry,
  			TaskManagerLocation.getHostName(remoteAddress),
  			resourceID,
  			taskManagerServicesConfiguration.getSystemResourceMetricsProbingInterval());
  
  		TaskManagerServices taskManagerServices = TaskManagerServices.fromConfiguration(
  			taskManagerServicesConfiguration,
  			taskManagerMetricGroup,
  			resourceID,
  			rpcService.getExecutor(), // TODO replace this later with some dedicated executor for io.
  			EnvironmentInformation.getSizeOfFreeHeapMemoryWithDefrag(),
  			EnvironmentInformation.getMaxJvmHeapMemory());
  
  		TaskManagerConfiguration taskManagerConfiguration = TaskManagerConfiguration.fromConfiguration(configuration);
  
  		String metricQueryServiceAddress = metricRegistry.getMetricQueryServiceGatewayRpcAddress();
  
  		return new TaskExecutor(
  			rpcService,
  			taskManagerConfiguration,
  			highAvailabilityServices,
  			taskManagerServices,
  			heartbeatServices,
  			taskManagerMetricGroup,
  			metricQueryServiceAddress,
  			blobCacheService,
  			fatalErrorHandler);
  	}
docs: --------------------------------------------------------------------------------------------