@Override
    protected void handleCommandWatchTopicList(CommandWatchTopicList commandWatchTopicList) {
        checkArgument(state == State.Connected);
        final long requestId = commandWatchTopicList.getRequestId();
        final long watcherId = commandWatchTopicList.getWatcherId();
        final NamespaceName namespaceName = NamespaceName.get(commandWatchTopicList.getNamespace());

        Pattern topicsPattern = Pattern.compile(commandWatchTopicList.hasTopicsPattern()
                ? commandWatchTopicList.getTopicsPattern() : TopicList.ALL_TOPICS_PATTERN);
        String topicsHash = commandWatchTopicList.hasTopicsHash()
                ? commandWatchTopicList.getTopicsHash() : null;

        final Semaphore lookupSemaphore = service.getLookupRequestSemaphore();
        if (lookupSemaphore.tryAcquire()) {
            isNamespaceOperationAllowed(namespaceName, NamespaceOperation.GET_TOPICS).thenApply(isAuthorized -> {
                if (isAuthorized) {
                    topicListService.handleWatchTopicList(namespaceName, watcherId, requestId, topicsPattern,
                            topicsHash, lookupSemaphore);
                } else {
                    final String msg = "Proxy Client is not authorized to watchTopicList";
                    log.warn("[{}] {} with role {} on namespace {}", remoteAddress, msg, getPrincipal(), namespaceName);
                    commandSender.sendErrorResponse(requestId, ServerError.AuthorizationError, msg);
                    lookupSemaphore.release();
                }
                return null;
            }).exceptionally(ex -> {
                logNamespaceNameAuthException(remoteAddress, "watchTopicList", getPrincipal(),
                        Optional.of(namespaceName), ex);
                final String msg = "Exception occurred while trying to handle command WatchTopicList";
                commandSender.sendErrorResponse(requestId, ServerError.AuthorizationError, msg);
                lookupSemaphore.release();
                return null;
            });
        } else {
            if (log.isDebugEnabled()) {
                log.debug("[{}] Failed WatchTopicList 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 shouldFailHandleCommandWatchTopicList() 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.handleCommandWatchTopicList(any());
    }