public ConsumeStatsList fetchConsumeStatsInBroker(String brokerAddr, boolean isOrder,
        long timeoutMillis) throws MQClientException,
        RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, InterruptedException {
        GetConsumeStatsInBrokerHeader requestHeader = new GetConsumeStatsInBrokerHeader();
        requestHeader.setIsOrder(isOrder);

        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_BROKER_CONSUME_STATS, requestHeader);
        RemotingCommand response = this.remotingClient
            .invokeSync(MixAll.brokerVIPChannel(this.clientConfig.isVipChannelEnabled(), brokerAddr), request, timeoutMillis);
        assert response != null;
        switch (response.getCode()) {
            case ResponseCode.SUCCESS: {
                byte[] body = response.getBody();
                if (body != null) {
                    return ConsumeStatsList.decode(body, ConsumeStatsList.class);
                }
            }
            default:
                break;
        }

        throw new MQClientException(response.getCode(), response.getRemark());
    }
@Test
    public void assertFetchConsumeStatsInBroker() throws RemotingException, InterruptedException, MQClientException {
        mockInvokeSync();
        ConsumeStatsList responseBody = new ConsumeStatsList();
        responseBody.setBrokerAddr(defaultBrokerAddr);
        responseBody.getConsumeStatsList().add(new HashMap<>());
        setResponseBody(responseBody);
        ConsumeStatsList actual = mqClientAPI.fetchConsumeStatsInBroker(defaultBrokerAddr, false, defaultTimeout);
        assertNotNull(actual);
        assertEquals(1, actual.getConsumeStatsList().size());
        assertEquals(defaultBrokerAddr, actual.getBrokerAddr());
    }