public Map<String, Map<MessageQueue, Long>> invokeBrokerToGetConsumerStatus(final String addr, final String topic,
        final String group,
        final String clientAddr,
        final long timeoutMillis) throws RemotingException, MQClientException, InterruptedException {
        GetConsumerStatusRequestHeader requestHeader = new GetConsumerStatusRequestHeader();
        requestHeader.setTopic(topic);
        requestHeader.setGroup(group);
        requestHeader.setClientAddr(clientAddr);

        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.INVOKE_BROKER_TO_GET_CONSUMER_STATUS, requestHeader);
        RemotingCommand response = this.remotingClient.invokeSync(MixAll.brokerVIPChannel(this.clientConfig.isVipChannelEnabled(), addr),
            request, timeoutMillis);
        assert response != null;
        switch (response.getCode()) {
            case ResponseCode.SUCCESS: {
                if (response.getBody() != null) {
                    GetConsumerStatusBody body = GetConsumerStatusBody.decode(response.getBody(), GetConsumerStatusBody.class);
                    return body.getConsumerTable();
                }
            }
            default:
                break;
        }

        throw new MQClientException(response.getCode(), response.getRemark());
    }
@Test
    public void assertInvokeBrokerToGetConsumerStatus() throws RemotingException, InterruptedException, MQClientException {
        mockInvokeSync();
        GetConsumerStatusBody responseBody = new GetConsumerStatusBody();
        responseBody.getConsumerTable().put("key", new HashMap<>());
        responseBody.getMessageQueueTable().put(new MessageQueue(), 1L);
        setResponseBody(responseBody);
        Map<String, Map<MessageQueue, Long>> actual = mqClientAPI.invokeBrokerToGetConsumerStatus(defaultBrokerAddr, defaultTopic, "", "", defaultTimeout);
        assertNotNull(actual);
        assertEquals(1, actual.size());
    }