@VisibleForTesting
    public static ActorSystem startRemoteActorSystem(
            Configuration configuration,
            String externalAddress,
            String externalPortRange,
            Logger logger)
            throws Exception {
        return startRemoteActorSystem(
                configuration,
                PekkoUtils.getFlinkActorSystemName(),
                externalAddress,
                externalPortRange,
                NetUtils.getWildcardIPAddress(),
                Optional.empty(),
                logger,
                PekkoUtils.getForkJoinExecutorConfig(
                        getForkJoinExecutorConfiguration(configuration)),
                null);
    }
@Test
    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 ActorSystemBootstrapTools
                                                                        .startRemoteActorSystem(
                                                                                new Configuration(),
                                                                                "localhost",
                                                                                "0",
                                                                                LOG);
                                                            }),
                                                    executorService))
                            .map(
                                    // terminate ActorSystems
                                    actorSystemFuture ->
                                            actorSystemFuture.thenCompose(
                                                    PekkoUtils::terminateActorSystem))
                            .collect(Collectors.toList());

            FutureUtils.completeAll(actorSystemFutures).get();
        } finally {
            ExecutorUtils.gracefulShutdown(10000L, TimeUnit.MILLISECONDS, executorService);
        }
    }