protected CompletableFuture<Optional<Topic>> loadOrCreatePersistentTopic(final String topic,
            boolean createIfMissing, Map<String, String> properties, @Nullable TopicPolicies topicPolicies) {
        final CompletableFuture<Optional<Topic>> topicFuture = FutureUtil.createFutureWithTimeout(
                Duration.ofSeconds(pulsar.getConfiguration().getTopicLoadTimeoutSeconds()), executor(),
                () -> FAILED_TO_LOAD_TOPIC_TIMEOUT_EXCEPTION);

        topicFuture.exceptionally(t -> {
            pulsarStats.recordTopicLoadFailed();
            return null;
        });

        checkTopicNsOwnership(topic)
                .thenRun(() -> {
                    final Semaphore topicLoadSemaphore = topicLoadRequestSemaphore.get();

                    if (topicLoadSemaphore.tryAcquire()) {
                        checkOwnershipAndCreatePersistentTopic(topic, createIfMissing, topicFuture,
                                properties, topicPolicies);
                        topicFuture.handle((persistentTopic, ex) -> {
                            // release permit and process pending topic
                            topicLoadSemaphore.release();
                            // do not recreate topic if topic is already migrated and deleted by broker
                            // so, avoid creating a new topic if migration is already started
                            if (ex != null && (ex.getCause() instanceof TopicMigratedException)) {
                                pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
                                topicFuture.completeExceptionally(ex.getCause());
                                return null;
                            }
                            createPendingLoadTopic();
                            return null;
                        });
                    } else {
                        pendingTopicLoadingQueue.add(new TopicLoadingContext(topic,
                                createIfMissing, topicFuture, properties, topicPolicies));
                        if (log.isDebugEnabled()) {
                            log.debug("topic-loading for {} added into pending queue", topic);
                        }
                    }
                }).exceptionally(ex -> {
                    pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
                    topicFuture.completeExceptionally(ex.getCause());
                    return null;
                });

        return topicFuture;
    }
@Test
    public void testConcurrentLoadTopicExceedLimitShouldNotBeAutoCreated() throws Exception {
        boolean needDeleteTopic = false;
        final String namespace = "prop/concurrentLoad";
        try {
            // set up broker disable auto create and set concurrent load to 1 qps.
            cleanup();
            conf.setMaxConcurrentTopicLoadRequest(1);
            conf.setAllowAutoTopicCreation(false);
            setup();

            try {
                admin.namespaces().createNamespace(namespace);
            } catch (PulsarAdminException.ConflictException e) {
                // Ok.. (if test fails intermittently and namespace is already created)
            }

            // create 3 topic
            String topicName = "persistent://" + namespace + "/my-topic";

            for (int i = 0; i < 3; i++) {
                admin.topics().createNonPartitionedTopic(topicName + "_" + i);
            }

            needDeleteTopic = true;

            // try to load 10 topic
            ArrayList<CompletableFuture<Optional<Topic>>> loadFutures = new ArrayList<>();
            for (int i = 0; i < 10; i++) {
                // try to create topic which should fail as bundle is disable
                CompletableFuture<Optional<Topic>> futureResult = pulsar.getBrokerService()
                        .loadOrCreatePersistentTopic(topicName + "_" + i, false, null, null);
                loadFutures.add(futureResult);
            }

            CompletableFuture<?>[] o = (CompletableFuture<?>[]) Array.newInstance(CompletableFuture.class, 10);
            CompletableFuture<?>[] completableFutures = loadFutures.toArray(o);
            CompletableFuture.allOf(completableFutures).get();

            // check topic load CompletableFuture. only first three topic should be success.
            for (int i = 0; i < 10; i++) {
                CompletableFuture<Optional<Topic>> load = loadFutures.get(i);
                if (i < 3) {
                    Assert.assertTrue(load.isDone());
                    Assert.assertFalse(load.isCompletedExceptionally());
                } else {
                    // check topic should not be created if disable autoCreateTopic.
                    Assert.assertTrue(load.isDone());
                    Assert.assertTrue(load.get().isEmpty());
                }
            }
        } finally {
            if (needDeleteTopic) {
                String topicName = "persistent://" + namespace + "/my-topic";

                for (int i = 0; i < 3; i++) {
                    admin.topics().delete(topicName + "_" + i);
                }
            }
        }
    }