@Test public void testConcurrentActorSystemCreation() throws Exception { final int concurrentCreations = 10; final ExecutorService executorService = Executors.newFixedThreadPool(concurrentCreations); final CyclicBarrier cyclicBarrier = new CyclicBarrier(concurrentCreations); try { final List<CompletableFuture<Void>> actorSystemFutures = IntStream.range(0, concurrentCreations) .mapToObj( ignored -> CompletableFuture.supplyAsync( CheckedSupplier.unchecked(() -> { cyclicBarrier.await(); return BootstrapTools.startActorSystem( new Configuration(), "localhost", "0", LOG); }), executorService)) .map( actorSystemFuture -> actorSystemFuture.thenCompose(AkkaUtils::terminateActorSystem) ).collect(Collectors.toList()); FutureUtils.completeAll(actorSystemFutures).get(); } finally { ExecutorUtils.gracefulShutdown(10000L, TimeUnit.MILLISECONDS, executorService); } }
public static ActorSystem startActorSystem( Configuration configuration, String listeningAddress, String portRangeDefinition, Logger logger) throws Exception { return startActorSystem( configuration, listeningAddress, portRangeDefinition, logger, ForkJoinExecutorConfiguration.fromConfiguration(configuration)); }
BootstrapTools { public static ActorSystem startActorSystem( Configuration configuration, String listeningAddress, String portRangeDefinition, Logger logger) throws Exception { return startActorSystem( configuration, listeningAddress, portRangeDefinition, logger, ForkJoinExecutorConfiguration.fromConfiguration(configuration)); } }
BootstrapTools { public static ActorSystem startActorSystem( Configuration configuration, String listeningAddress, String portRangeDefinition, Logger logger) throws Exception { return startActorSystem( configuration, listeningAddress, portRangeDefinition, logger, ForkJoinExecutorConfiguration.fromConfiguration(configuration)); } private  BootstrapTools(); }
BootstrapTools { public static ActorSystem startActorSystem( Configuration configuration, String listeningAddress, String portRangeDefinition, Logger logger) throws Exception { return startActorSystem( configuration, listeningAddress, portRangeDefinition, logger, ForkJoinExecutorConfiguration.fromConfiguration(configuration)); } private  BootstrapTools(); static ActorSystem startActorSystem(
		Configuration configuration,
		String listeningAddress,
		String portRangeDefinition,
		Logger logger); static ActorSystem startActorSystem(
			Configuration configuration,
			String listeningAddress,
			String portRangeDefinition,
			Logger logger,
			@Nonnull ActorSystemExecutorConfiguration actorSystemExecutorConfiguration); static ActorSystem startActorSystem(
			Configuration configuration,
			String actorSystemName,
			String listeningAddress,
			String portRangeDefinition,
			Logger logger,
			@Nonnull ActorSystemExecutorConfiguration actorSystemExecutorConfiguration); static ActorSystem startActorSystem(
		Configuration configuration,
		String listeningAddress,
		int listeningPort,
		Logger logger); static ActorSystem startActorSystem(
				Configuration configuration,
				String listeningAddress,
				int listeningPort,
				Logger logger,
				ActorSystemExecutorConfiguration actorSystemExecutorConfiguration); static ActorSystem startActorSystem(
		Configuration configuration,
		String actorSystemName,
		String listeningAddress,
		int listeningPort,
		Logger logger,
		ActorSystemExecutorConfiguration actorSystemExecutorConfiguration); static Configuration generateTaskManagerConfiguration(
				Configuration baseConfig,
				String jobManagerHostname,
				int jobManagerPort,
				int numSlots,
				FiniteDuration registrationTimeout); static void writeConfiguration(Configuration cfg, File file); static void substituteDeprecatedConfigKey(Configuration config, String deprecated, String designated); static void substituteDeprecatedConfigPrefix(
			Configuration config,
			String deprecatedPrefix,
			String designatedPrefix); static Option newDynamicPropertiesOption(); static Configuration parseDynamicProperties(CommandLine cmd); static String getTaskManagerShellCommand(
			Configuration flinkConfig,
			ContaineredTaskManagerParameters tmParams,
			String configDirectory,
			String logDirectory,
			boolean hasLogback,
			boolean hasLog4j,
			boolean hasKrb5,
			Class<?> mainClass); static String getStartCommand(String template,
		Map<String, String> startCommandValues); static void updateTmpDirectoriesInConfiguration(
			Configuration configuration,
			@Nullable String defaultDirs); static Configuration cloneConfiguration(Configuration configuration); }
BootstrapTools { public static ActorSystem startActorSystem( Configuration configuration, String listeningAddress, String portRangeDefinition, Logger logger) throws Exception { return startActorSystem( configuration, listeningAddress, portRangeDefinition, logger, ForkJoinExecutorConfiguration.fromConfiguration(configuration)); } private  BootstrapTools(); static ActorSystem startActorSystem(
		Configuration configuration,
		String listeningAddress,
		String portRangeDefinition,
		Logger logger); static ActorSystem startActorSystem(
			Configuration configuration,
			String listeningAddress,
			String portRangeDefinition,
			Logger logger,
			@Nonnull ActorSystemExecutorConfiguration actorSystemExecutorConfiguration); static ActorSystem startActorSystem(
			Configuration configuration,
			String actorSystemName,
			String listeningAddress,
			String portRangeDefinition,
			Logger logger,
			@Nonnull ActorSystemExecutorConfiguration actorSystemExecutorConfiguration); static ActorSystem startActorSystem(
		Configuration configuration,
		String listeningAddress,
		int listeningPort,
		Logger logger); static ActorSystem startActorSystem(
				Configuration configuration,
				String listeningAddress,
				int listeningPort,
				Logger logger,
				ActorSystemExecutorConfiguration actorSystemExecutorConfiguration); static ActorSystem startActorSystem(
		Configuration configuration,
		String actorSystemName,
		String listeningAddress,
		int listeningPort,
		Logger logger,
		ActorSystemExecutorConfiguration actorSystemExecutorConfiguration); static Configuration generateTaskManagerConfiguration(
				Configuration baseConfig,
				String jobManagerHostname,
				int jobManagerPort,
				int numSlots,
				FiniteDuration registrationTimeout); static void writeConfiguration(Configuration cfg, File file); static void substituteDeprecatedConfigKey(Configuration config, String deprecated, String designated); static void substituteDeprecatedConfigPrefix(
			Configuration config,
			String deprecatedPrefix,
			String designatedPrefix); static Option newDynamicPropertiesOption(); static Configuration parseDynamicProperties(CommandLine cmd); static String getTaskManagerShellCommand(
			Configuration flinkConfig,
			ContaineredTaskManagerParameters tmParams,
			String configDirectory,
			String logDirectory,
			boolean hasLogback,
			boolean hasLog4j,
			boolean hasKrb5,
			Class<?> mainClass); static String getStartCommand(String template,
		Map<String, String> startCommandValues); static void updateTmpDirectoriesInConfiguration(
			Configuration configuration,
			@Nullable String defaultDirs); static Configuration cloneConfiguration(Configuration configuration);  }