code: 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);
  		}
  	}
docs: 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