private RemotingCommand getConsumerRunningInfo(ChannelHandlerContext ctx,
        RemotingCommand request) throws RemotingCommandException {
        final GetConsumerRunningInfoRequestHeader requestHeader =
            (GetConsumerRunningInfoRequestHeader) request.decodeCommandCustomHeader(GetConsumerRunningInfoRequestHeader.class);

        return this.callConsumer(RequestCode.GET_CONSUMER_RUNNING_INFO, request, requestHeader.getConsumerGroup(),
            requestHeader.getClientId());
    }
@Test
    public void testGetConsumerRunningInfo() throws RemotingCommandException, RemotingSendRequestException, RemotingTimeoutException, InterruptedException {
        when(brokerController.getConsumerManager()).thenReturn(consumerManager);
        when(consumerManager.findChannel(anyString(),anyString())).thenReturn(null);
        GetConsumerRunningInfoRequestHeader requestHeader = new GetConsumerRunningInfoRequestHeader();
        requestHeader.setClientId("client");
        requestHeader.setConsumerGroup("group");
        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.GET_CONSUMER_RUNNING_INFO, requestHeader);
        request.makeCustomHeaderToNet();
        RemotingCommand response = adminBrokerProcessor.processRequest(handlerContext, request);
        assertThat(response.getCode()).isEqualTo(ResponseCode.SYSTEM_ERROR);

        when(consumerManager.findChannel(anyString(),anyString())).thenReturn(clientChannelInfo);
        when(clientChannelInfo.getVersion()).thenReturn(MQVersion.Version.V3_0_0_SNAPSHOT.ordinal());
        response = adminBrokerProcessor.processRequest(handlerContext, request);
        assertThat(response.getCode()).isEqualTo(ResponseCode.SYSTEM_ERROR);

        when(clientChannelInfo.getVersion()).thenReturn(MQVersion.Version.V5_2_3.ordinal());
        when(brokerController.getBroker2Client()).thenReturn(broker2Client);
        when(clientChannelInfo.getChannel()).thenReturn(channel);
        RemotingCommand responseCommand = RemotingCommand.createResponseCommand(null);
        responseCommand.setCode(ResponseCode.SUCCESS);
        when(broker2Client.callClient(any(Channel.class),any(RemotingCommand.class))).thenReturn(responseCommand);
        response = adminBrokerProcessor.processRequest(handlerContext, request);
        assertThat(response.getCode()).isEqualTo(ResponseCode.SUCCESS);

        when(broker2Client.callClient(any(Channel.class),any(RemotingCommand.class))).thenThrow(new RemotingTimeoutException("timeout"));
        response = adminBrokerProcessor.processRequest(handlerContext, request);
        assertThat(response.getCode()).isEqualTo(ResponseCode.CONSUME_MSG_TIMEOUT);
    }