@Override
    protected void handleGetTopicsOfNamespace(CommandGetTopicsOfNamespace commandGetTopicsOfNamespace) {
        checkArgument(state == State.Connected);
        final long requestId = commandGetTopicsOfNamespace.getRequestId();
        final String namespace = commandGetTopicsOfNamespace.getNamespace();
        final CommandGetTopicsOfNamespace.Mode mode = commandGetTopicsOfNamespace.getMode();
        final Optional<String> topicsPattern = Optional.ofNullable(commandGetTopicsOfNamespace.hasTopicsPattern()
                ? commandGetTopicsOfNamespace.getTopicsPattern() : null);
        final Optional<String> topicsHash = Optional.ofNullable(commandGetTopicsOfNamespace.hasTopicsHash()
                ? commandGetTopicsOfNamespace.getTopicsHash() : null);
        final NamespaceName namespaceName = NamespaceName.get(namespace);

        final Semaphore lookupSemaphore = service.getLookupRequestSemaphore();
        if (lookupSemaphore.tryAcquire()) {
            isNamespaceOperationAllowed(namespaceName, NamespaceOperation.GET_TOPICS).thenApply(isAuthorized -> {
                if (isAuthorized) {
                    getBrokerService().pulsar().getNamespaceService().getListOfUserTopics(namespaceName, mode)
                        .thenAccept(topics -> {
                            boolean filterTopics = false;
                            // filter system topic
                            List<String> filteredTopics = topics;

                            if (enableSubscriptionPatternEvaluation && topicsPattern.isPresent()) {
                                if (topicsPattern.get().length() <= maxSubscriptionPatternLength) {
                                    filterTopics = true;
                                    filteredTopics = TopicList.filterTopics(filteredTopics, topicsPattern.get());
                                } else {
                                    log.info("[{}] Subscription pattern provided was longer than maximum {}.",
                                            remoteAddress, maxSubscriptionPatternLength);
                                }
                            }
                            String hash = TopicList.calculateHash(filteredTopics);
                            boolean hashUnchanged = topicsHash.isPresent() && topicsHash.get().equals(hash);
                            if (hashUnchanged) {
                                filteredTopics = Collections.emptyList();
                            }
                            if (log.isDebugEnabled()) {
                                log.debug(
                                        "[{}] Received CommandGetTopicsOfNamespace for namespace [//{}] by {}, size:{}",
                                        remoteAddress, namespace, requestId, topics.size());
                            }
                            commandSender.sendGetTopicsOfNamespaceResponse(filteredTopics, hash, filterTopics,
                                    !hashUnchanged, requestId);
                            lookupSemaphore.release();
                        })
                        .exceptionally(ex -> {
                            log.warn("[{}] Error GetTopicsOfNamespace for namespace [//{}] by {}",
                                    remoteAddress, namespace, requestId);
                            commandSender.sendErrorResponse(requestId,
                                    BrokerServiceException.getClientErrorCode(new ServerMetadataException(ex)),
                                    ex.getMessage());
                            lookupSemaphore.release();
                            return null;
                        });
                } else {
                    final String msg = "Client is not authorized to GetTopicsOfNamespace";
                    log.warn("[{}] {} with role {} on namespace {}", remoteAddress, msg, getPrincipal(), namespaceName);
                    commandSender.sendErrorResponse(requestId, ServerError.AuthorizationError, msg);
                    lookupSemaphore.release();
                }
                return null;
            }).exceptionally(ex -> {
                logNamespaceNameAuthException(remoteAddress, "GetTopicsOfNamespace", getPrincipal(),
                        Optional.of(namespaceName), ex);
                final String msg = "Exception occurred while trying to authorize GetTopicsOfNamespace";
                commandSender.sendErrorResponse(requestId, ServerError.AuthorizationError, msg);
                lookupSemaphore.release();
                return null;
            });
        } else {
            if (log.isDebugEnabled()) {
                log.debug("[{}] Failed GetTopicsOfNamespace lookup due to too many lookup-requests {}", remoteAddress,
                        namespaceName);
            }
            commandSender.sendErrorResponse(requestId, ServerError.TooManyRequests,
                    "Failed due to too many pending lookup requests");
        }
    }
@Test(expectedExceptions = IllegalArgumentException.class)
    public void shouldFailHandleGetTopicsOfNamespace() throws Exception {
        ServerCnx serverCnx = mock(ServerCnx.class, CALLS_REAL_METHODS);
        Field stateUpdater = ServerCnx.class.getDeclaredField("state");
        stateUpdater.setAccessible(true);
        stateUpdater.set(serverCnx, ServerCnx.State.Failed);
        serverCnx.handleGetTopicsOfNamespace(any());
    }