flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/main/java/org/apache/flink/runtime/rpc/akka/AkkaRpcServiceUtils.java
/**
	 * Utility method to create RPC service from configuration and hostname, port.
	 *
	 * @param hostname   The hostname/address that describes the TaskManager's data location.
	 * @param port           If true, the TaskManager will not initiate the TCP network stack.
	 * @param configuration                 The configuration for the TaskManager.
	 * @return   The rpc service which is used to start and connect to the TaskManager RpcEndpoint .
	 * @throws IOException      Thrown, if the actor system can not bind to the address
	 * @throws Exception      Thrown is some other error occurs while creating akka actor system
	 */
Utility method to create RPC service from configuration and hostname, port.
@param hostname   The hostname/address that describes the TaskManager's data location.
@param port           If true, the TaskManager will not initiate the TCP network stack.
@param configuration                 The configuration for the TaskManager.
@return   The rpc service which is used to start and connect to the TaskManager RpcEndpoint .
@throws IOException      Thrown, if the actor system can not bind to the address
@throws Exception      Thrown is some other error occurs while creating akka actor system
Utility method to create RPC service from configuration and hostname port . @param hostname The hostname / address that describes the TaskManager ' s data location . @param port If true the TaskManager will not initiate the TCP network stack . @param configuration The configuration for the TaskManager . @return The rpc service which is used to start and connect to the TaskManager RpcEndpoint . @throws IOException Thrown if the actor system can not bind to the address @throws Exception Thrown is some other error occurs while creating akka actor system
public static RpcService createRpcService(String hostname, int port, Configuration configuration) throws Exception {
		LOG.info("Starting AkkaRpcService at {}.", NetUtils.unresolvedHostAndPortToNormalizedString(hostname, port));
		final ActorSystem actorSystem;
		try {
			Config akkaConfig;
			if (hostname != null && !hostname.isEmpty()) {
				akkaConfig = AkkaUtils.getAkkaConfig(configuration, hostname, port);
			} else {
				akkaConfig = AkkaUtils.getAkkaConfig(configuration);
			}
			LOG.debug("Using akka configuration \n {}.", akkaConfig);
			actorSystem = AkkaUtils.createActorSystem(akkaConfig);
		} catch (Throwable t) {
			if (t instanceof ChannelException) {
				Throwable cause = t.getCause();
				if (cause != null && t.getCause() instanceof java.net.BindException) {
					String address = NetUtils.hostAndPortToUrlString(hostname, port);
					throw new IOException("Unable to bind AkkaRpcService actor system to address " +
						address + " - " + cause.getMessage(), t);
				}
			}
			throw new Exception("Could not create TaskManager actor system", t);
		}
		final Time timeout = Time.milliseconds(AkkaUtils.getTimeout(configuration).toMillis());
		return new AkkaRpcService(actorSystem, timeout);
	}
public static RpcService createRpcService ( String hostname , int port , Configuration configuration ) throws Exception { LOG . info ( "Starting AkkaRpcService at {}." , NetUtils . unresolvedHostAndPortToNormalizedString ( hostname , port ) ) ; final ActorSystem actorSystem ; try { Config akkaConfig ; if ( hostname != null && ! hostname . isEmpty ( ) ) { akkaConfig = AkkaUtils . getAkkaConfig ( configuration , hostname , port ) ; } else { akkaConfig = AkkaUtils . getAkkaConfig ( configuration ) ; } LOG . debug ( "Using akka configuration \n {}." , akkaConfig ) ; actorSystem = AkkaUtils . createActorSystem ( akkaConfig ) ; } catch ( Throwable t ) { if ( t instanceof ChannelException ) { Throwable cause = t . getCause ( ) ; if ( cause != null && t . getCause ( ) instanceof java . net . BindException ) { String address = NetUtils . hostAndPortToUrlString ( hostname , port ) ; throw new IOException ( "Unable to bind AkkaRpcService actor system to address " + address + " - " + cause . getMessage ( ) , t ) ; } } throw new Exception ( "Could not create TaskManager actor system" , t ) ; } final Time timeout = Time . milliseconds ( AkkaUtils . getTimeout ( configuration ) . toMillis ( ) ) ; return new AkkaRpcService ( actorSystem , timeout ) ; }
Utility method to create RPC service from configuration and hostname, port.
Utility method to create RPC service from configuration and hostname port .
// remote akka config // local akka config