@Override
    protected void handleConsumerStats(CommandConsumerStats commandConsumerStats) {
        checkArgument(state == State.Connected);
        if (log.isDebugEnabled()) {
            log.debug("Received CommandConsumerStats call from {}", remoteAddress);
        }

        final long requestId = commandConsumerStats.getRequestId();
        final long consumerId = commandConsumerStats.getConsumerId();
        CompletableFuture<Consumer> consumerFuture = consumers.get(consumerId);
        Consumer consumer = consumerFuture.getNow(null);
        ByteBuf msg = null;

        if (consumer == null) {
            log.error(
                    "Failed to get consumer-stats response - Consumer not found for"
                            + " CommandConsumerStats[remoteAddress = {}, requestId = {}, consumerId = {}]",
                    remoteAddress, requestId, consumerId);
            msg = Commands.newConsumerStatsResponse(ServerError.ConsumerNotFound,
                    "Consumer " + consumerId + " not found", requestId);
        } else {
            if (log.isDebugEnabled()) {
                log.debug("CommandConsumerStats[requestId = {}, consumer = {}]", requestId, consumer);
            }
            msg = createConsumerStatsResponse(consumer, requestId);
        }

        writeAndFlush(msg);
    }
@Test(expectedExceptions = IllegalArgumentException.class)
    public void shouldFailHandleConsumerStats() 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.handleConsumerStats(any());
    }