public CompletableFuture<Void> subscribeAsync(String topicName, boolean createTopicIfDoesNotExist) {
        TopicName topicNameInstance = getTopicName(topicName);
        if (topicNameInstance == null) {
            return FutureUtil.failedFuture(
                    new PulsarClientException.AlreadyClosedException("Topic name not valid"));
        }
        String fullTopicName = topicNameInstance.toString();
        if (consumers.containsKey(fullTopicName)
                || partitionedTopics.containsKey(topicNameInstance.getPartitionedTopicName())) {
            return FutureUtil.failedFuture(
                    new PulsarClientException.AlreadyClosedException("Already subscribed to " + topicName));
        }

        if (getState() == State.Closing || getState() == State.Closed) {
            return FutureUtil.failedFuture(
                new PulsarClientException.AlreadyClosedException("Topics Consumer was already closed"));
        }

        CompletableFuture<Void> subscribeResult = new CompletableFuture<>();

        client.getPartitionedTopicMetadata(topicName, true, false)
                .thenAccept(metadata -> subscribeTopicPartitions(subscribeResult, fullTopicName, metadata.partitions,
                    createTopicIfDoesNotExist))
                .exceptionally(ex1 -> {
                    log.warn("[{}] Failed to get partitioned topic metadata: {}", fullTopicName, ex1.getMessage());
                    subscribeResult.completeExceptionally(ex1);
                    return null;
                });

        return subscribeResult;
    }
@Test(timeOut = 10000)
    public void testParallelSubscribeAsync() throws Exception {
        String topicName = "parallel-subscribe-async-topic";
        MultiTopicsConsumerImpl<byte[]> impl = createMultiTopicsConsumer();

        CompletableFuture<Void> firstInvocation = impl.subscribeAsync(topicName, true);
        Thread.sleep(5); // less than completionDelayMillis
        CompletableFuture<Void> secondInvocation = impl.subscribeAsync(topicName, true);

        firstInvocation.get(); // does not throw
        Throwable t = expectThrows(ExecutionException.class, secondInvocation::get);
        Throwable cause = t.getCause();
        assertEquals(cause.getClass(), PulsarClientException.class);
        assertTrue(cause.getMessage().endsWith("Topic is already being subscribed for in other thread."));
    }