public void handleWatchTopicList(NamespaceName namespaceName, long watcherId, long requestId, Pattern topicsPattern,
                                     String topicsHash, Semaphore lookupSemaphore) {

        if (!enableSubscriptionPatternEvaluation || topicsPattern.pattern().length() > maxSubscriptionPatternLength) {
            String msg = "Unable to create topic list watcher: ";
            if (!enableSubscriptionPatternEvaluation) {
                msg += "Evaluating subscription patterns is disabled.";
            } else {
                msg += "Pattern longer than maximum: " + maxSubscriptionPatternLength;
            }
            log.warn("[{}] {} on namespace {}", connection.toString(), msg, namespaceName);
            connection.getCommandSender().sendErrorResponse(requestId, ServerError.NotAllowedError, msg);
            lookupSemaphore.release();
            return;
        }
        CompletableFuture<TopicListWatcher> watcherFuture = new CompletableFuture<>();
        CompletableFuture<TopicListWatcher> existingWatcherFuture = watchers.putIfAbsent(watcherId, watcherFuture);

        if (existingWatcherFuture != null) {
            if (existingWatcherFuture.isDone() && !existingWatcherFuture.isCompletedExceptionally()) {
                TopicListWatcher watcher = existingWatcherFuture.getNow(null);
                log.info("[{}] Watcher with the same id is already created:"
                                + " watcherId={}, watcher={}",
                        connection.toString(), watcherId, watcher);
                watcherFuture = existingWatcherFuture;
            } else {
                // There was an early request to create a watcher with the same watcherId. This can happen when
                // client timeout is lower the broker timeouts. We need to wait until the previous watcher
                // creation request either completes or fails.
                log.warn("[{}] Watcher with id is already present on the connection,"
                        + " consumerId={}", connection.toString(), watcherId);
                ServerError error;
                if (!existingWatcherFuture.isDone()) {
                    error = ServerError.ServiceNotReady;
                } else {
                    error = ServerError.UnknownError;
                    watchers.remove(watcherId, existingWatcherFuture);
                }
                connection.getCommandSender().sendErrorResponse(requestId, error,
                        "Topic list watcher is already present on the connection");
                lookupSemaphore.release();
                return;
            }
        } else {
            initializeTopicsListWatcher(watcherFuture, namespaceName, watcherId, topicsPattern);
        }


        CompletableFuture<TopicListWatcher> finalWatcherFuture = watcherFuture;
        finalWatcherFuture.thenAccept(watcher -> {
                    List<String> topicList = watcher.getMatchingTopics();
                    String hash = TopicList.calculateHash(topicList);
                    if (hash.equals(topicsHash)) {
                        topicList = Collections.emptyList();
                    }
                    if (log.isDebugEnabled()) {
                        log.debug(
                                "[{}] Received WatchTopicList for namespace [//{}] by {}",
                                connection.toString(), namespaceName, requestId);
                    }
                    connection.getCommandSender().sendWatchTopicListSuccess(requestId, watcherId, hash, topicList);
                    lookupSemaphore.release();
                })
                .exceptionally(ex -> {
                    log.warn("[{}] Error WatchTopicList for namespace [//{}] by {}",
                            connection.toString(), namespaceName, requestId);
                    connection.getCommandSender().sendErrorResponse(requestId,
                            BrokerServiceException.getClientErrorCode(
                                    new BrokerServiceException.ServerMetadataException(ex)), ex.getMessage());
                    watchers.remove(watcherId, finalWatcherFuture);
                    lookupSemaphore.release();
                    return null;
                });
    }
@Test
    public void testCommandWatchErrorResponse() {
        topicListService.handleWatchTopicList(
                NamespaceName.get("tenant/ns"),
                13,
                7,
                Pattern.compile("persistent://tenant/ns/topic\\d"),
                null,
                lookupSemaphore);
        topicListFuture.completeExceptionally(new PulsarServerException("Error"));
        Assert.assertEquals(1, lookupSemaphore.availablePermits());
        verifyNoInteractions(topicResources);
        verify(connection.getCommandSender()).sendErrorResponse(eq(7L), any(ServerError.class),
                eq(PulsarServerException.class.getCanonicalName() + ": Error"));
    }