flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskManagerRunner.java
/**
	 * Create a RPC service for the task manager.
	 *
	 * @param configuration The configuration for the TaskManager.
	 * @param haServices to use for the task manager hostname retrieval
	 */
Create a RPC service for the task manager.
@param configuration The configuration for the TaskManager.
@param haServices to use for the task manager hostname retrieval
Create a RPC service for the task manager . @param configuration The configuration for the TaskManager . @param haServices to use for the task manager hostname retrieval
public static RpcService createRpcService(
		final Configuration configuration,
		final HighAvailabilityServices haServices) throws Exception {
		checkNotNull(configuration);
		checkNotNull(haServices);
		String taskManagerHostname = configuration.getString(ConfigConstants.TASK_MANAGER_HOSTNAME_KEY, null);
		if (taskManagerHostname != null) {
			LOG.info("Using configured hostname/address for TaskManager: {}.", taskManagerHostname);
		} else {
			Time lookupTimeout = Time.milliseconds(AkkaUtils.getLookupTimeout(configuration).toMillis());
			InetAddress taskManagerAddress = LeaderRetrievalUtils.findConnectingAddress(
				haServices.getResourceManagerLeaderRetriever(),
				lookupTimeout);
			taskManagerHostname = taskManagerAddress.getHostName();
			LOG.info("TaskManager will use hostname/address '{}' ({}) for communication.",
				taskManagerHostname, taskManagerAddress.getHostAddress());
		}
		final int rpcPort = configuration.getInteger(ConfigConstants.TASK_MANAGER_IPC_PORT_KEY, 0);
		Preconditions.checkState(rpcPort >= 0 && rpcPort <= 65535, "Invalid value for " +
				"'%s' (port for the TaskManager actor system) : %d - Leave config parameter empty or " +
				"use 0 to let the system choose port automatically.",
			ConfigConstants.TASK_MANAGER_IPC_PORT_KEY, rpcPort);
		return AkkaRpcServiceUtils.createRpcService(taskManagerHostname, rpcPort, configuration);
	}
public static RpcService createRpcService ( final Configuration configuration , final HighAvailabilityServices haServices ) throws Exception { checkNotNull ( configuration ) ; checkNotNull ( haServices ) ; String taskManagerHostname = configuration . getString ( ConfigConstants . TASK_MANAGER_HOSTNAME_KEY , null ) ; if ( taskManagerHostname != null ) { LOG . info ( "Using configured hostname/address for TaskManager: {}." , taskManagerHostname ) ; } else { Time lookupTimeout = Time . milliseconds ( AkkaUtils . getLookupTimeout ( configuration ) . toMillis ( ) ) ; InetAddress taskManagerAddress = LeaderRetrievalUtils . findConnectingAddress ( haServices . getResourceManagerLeaderRetriever ( ) , lookupTimeout ) ; taskManagerHostname = taskManagerAddress . getHostName ( ) ; LOG . info ( "TaskManager will use hostname/address '{}' ({}) for communication." , taskManagerHostname , taskManagerAddress . getHostAddress ( ) ) ; } final int rpcPort = configuration . getInteger ( ConfigConstants . TASK_MANAGER_IPC_PORT_KEY , 0 ) ; Preconditions . checkState ( rpcPort >= 0 && rpcPort <= 65535 , "Invalid value for " + "'%s' (port for the TaskManager actor system) : %d - Leave config parameter empty or " + "use 0 to let the system choose port automatically." , ConfigConstants . TASK_MANAGER_IPC_PORT_KEY , rpcPort ) ; return AkkaRpcServiceUtils . createRpcService ( taskManagerHostname , rpcPort , configuration ) ; }
Create a RPC service for the task manager.
Create a RPC service for the task manager .