public static ActorSystem startActorSystem(
		Configuration configuration,
		String actorSystemName,
		String listeningAddress,
		int listeningPort,
		Logger logger,
		ActorSystemExecutorConfiguration actorSystemExecutorConfiguration) throws Exception {

		String hostPortUrl = NetUtils.unresolvedHostAndPortToNormalizedString(listeningAddress, listeningPort);
		logger.info("Trying to start actor system at {}", hostPortUrl);

		try {
			Config akkaConfig = AkkaUtils.getAkkaConfig(
				configuration,
				new Some<>(new Tuple2<>(listeningAddress, listeningPort)),
				actorSystemExecutorConfiguration.getAkkaConfig());

			logger.debug("Using akka configuration\n {}", akkaConfig);

			ActorSystem actorSystem = AkkaUtils.createActorSystem(actorSystemName, akkaConfig);

			logger.info("Actor system started at {}", AkkaUtils.getAddress(actorSystem));
			return actorSystem;
		}
		catch (Throwable t) {
			if (t instanceof ChannelException) {
				Throwable cause = t.getCause();
				if (cause != null && t.getCause() instanceof BindException) {
					throw new IOException("Unable to create ActorSystem at address " + hostPortUrl +
						" : " + cause.getMessage(), t);
				}
			}
			throw new Exception("Could not create actor system", t);
		}
	}
public static ActorSystem startActorSystem ( Configuration configuration , String actorSystemName , String listeningAddress , int listeningPort , Logger logger , ActorSystemExecutorConfiguration actorSystemExecutorConfiguration ) throws Exception { String hostPortUrl = NetUtils . unresolvedHostAndPortToNormalizedString ( listeningAddress , listeningPort ) ; logger . info ( "Trying to start actor system at {}" , hostPortUrl ) ; try { Config akkaConfig = AkkaUtils . getAkkaConfig ( configuration , new Some < > ( new Tuple2 < > ( listeningAddress , listeningPort ) ) , actorSystemExecutorConfiguration . getAkkaConfig ( ) ) ; logger . debug ( "Using akka configuration\n {}" , akkaConfig ) ; ActorSystem actorSystem = AkkaUtils . createActorSystem ( actorSystemName , akkaConfig ) ; logger . info ( "Actor system started at {}" , AkkaUtils . getAddress ( actorSystem ) ) ; return actorSystem ; } catch ( Throwable t ) { if ( t instanceof ChannelException ) { Throwable cause = t . getCause ( ) ; if ( cause != null && t . getCause ( ) instanceof BindException ) { throw new IOException ( "Unable to create ActorSystem at address " + hostPortUrl + " : " + cause . getMessage ( ) , t ) ; } } throw new Exception ( "Could not create actor system" , t ) ; } }
Starts an Actor System at a specific port.
@param configuration The Flink configuration.
@param actorSystemName Name of the started {@link ActorSystem}
@param listeningAddress The address to listen at.
@param listeningPort The port to listen at.
@param logger the logger to output log information.
@param actorSystemExecutorConfiguration configuration for the ActorSystem's underlying executor
@return The ActorSystem which has been started.
@throws Exception