### Question:
DeFiBusClientAPIImpl extends MQClientAPIImpl { public List<String> getConsumerIdListByGroupAndTopic( final String addr, final String consumerGroup, final String topic, final long timeoutMillis) throws RemotingConnectException, RemotingSendRequestException, RemotingTimeoutException, MQBrokerException, InterruptedException { GetConsumerListByGroupAndTopicRequestHeader requestHeader = new GetConsumerListByGroupAndTopicRequestHeader(); requestHeader.setConsumerGroup(consumerGroup); requestHeader.setTopic(topic); RemotingCommand request = RemotingCommand.createRequestCommand(DeFiBusRequestCode.GET_CONSUMER_LIST_BY_GROUP_AND_TOPIC, requestHeader); RemotingCommand response = this.getRemotingClient().invokeSync(MixAll.brokerVIPChannel(false, addr), request, timeoutMillis); assert response != null; switch (response.getCode()) { case ResponseCode.SUCCESS: { if (response.getBody() != null) { GetConsumerListByGroupAndTopicResponseBody body = GetConsumerListByGroupAndTopicResponseBody.decode(response.getBody(), GetConsumerListByGroupAndTopicResponseBody.class); return body.getConsumerIdList(); } } default: break; } throw new MQBrokerException(response.getCode(), response.getRemark()); } DeFiBusClientAPIImpl(NettyClientConfig nettyClientConfig,
        ClientRemotingProcessor clientRemotingProcessor,
        RPCHook rpcHook,
        ClientConfig clientConfig); void setWsAddr(String wsAddr); @Override SendResult sendMessage(//
        final String addr, // 1
        final String brokerName, // 2
        final Message msg, // 3
        final SendMessageRequestHeader requestHeader, // 4
        final long timeoutMillis, // 5
        final CommunicationMode communicationMode, // 6
        final SendCallback sendCallback, // 7
        final TopicPublishInfo topicPublishInfo, // 8
        final MQClientInstance instance, // 9
        final int retryTimesWhenSendFailed, // 10
        final SendMessageContext context, // 11
        final DefaultMQProducerImpl producer // 12
    ); List<String> getConsumerIdListByGroupAndTopic(
        final String addr,
        final String consumerGroup,
        final String topic,
        final long timeoutMillis); static final Logger LOGGER; }### Answer:
@Test public void testGetConsumerIdListByGroupAndTopic_Success() throws Exception { doAnswer(new Answer() { @Override public Object answer(InvocationOnMock mock) throws Throwable { RemotingCommand request = mock.getArgument(1); GetConsumerListByGroupResponseBody body = new GetConsumerListByGroupResponseBody(); List<String> cidList = new ArrayList<>(); for (int i = 0; i < 10; i++) { cidList.add(i + ""); } body.setConsumerIdList(cidList); request.setBody(body.encode()); request.setCode(ResponseCode.SUCCESS); request.setOpaque(request.getOpaque()); return request; } }).when(remotingClient).invokeSync(anyString(), any(RemotingCommand.class), anyLong()); List<String> list = deFiBusClientAPI.getConsumerIdListByGroupAndTopic(brokerAddr, group, topic, 3 * 1000); assertThat(list.size()).isEqualTo(10); }
@Test(expected = MQBrokerException.class) public void testGetConsumerIdListByGroupAndTopic_OnException() throws Exception { doAnswer(new Answer() { @Override public Object answer(InvocationOnMock mock) throws Throwable { RemotingCommand request = mock.getArgument(1); request.setCode(ResponseCode.SYSTEM_ERROR); return request; } }).when(remotingClient).invokeSync(anyString(), any(RemotingCommand.class), anyLong()); deFiBusClientAPI.getConsumerIdListByGroupAndTopic(brokerAddr, group, topic, 3 * 1000); }