#490 向指定Consumer发起消费消息请求，运维需要。服务端开发完成。

Author: vintagewang

rocketmq-broker/src/main/java/com/alibaba/rocketmq/broker/client/net/Broker2Client.java

@@ -108,7 +108,7 @@ public RemotingCommand getConsumerRunningInfo(//
     }
 
 
-    public RemotingCommand consumeMessageDirectly(//
+    public RemotingCommand callClient(//
             final Channel channel,//
             final RemotingCommand request//
     ) throws RemotingSendRequestException, RemotingTimeoutException, InterruptedException {

rocketmq-broker/src/main/java/com/alibaba/rocketmq/broker/processor/AdminBrokerProcessor.java

@@ -90,6 +90,7 @@
 import com.alibaba.rocketmq.common.protocol.header.filtersrv.RegisterFilterServerResponseHeader;
 import com.alibaba.rocketmq.common.protocol.heartbeat.SubscriptionData;
 import com.alibaba.rocketmq.common.subscription.SubscriptionGroupConfig;
+import com.alibaba.rocketmq.remoting.CommandCustomHeader;
 import com.alibaba.rocketmq.remoting.common.RemotingHelper;
 import com.alibaba.rocketmq.remoting.exception.RemotingCommandException;
 import com.alibaba.rocketmq.remoting.netty.NettyRequestProcessor;
@@ -228,93 +229,67 @@ public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand
     }
 
 
-    private RemotingCommand consumeMessageDirectly(ChannelHandlerContext ctx, RemotingCommand request)
-            throws RemotingCommandException {
+    private RemotingCommand callConsumer(//
+            final int requestCode,//
+            final CommandCustomHeader requestHeader, //
+            final String consumerGroup,//
+            final String clientId) throws RemotingCommandException {
         final RemotingCommand response = RemotingCommand.createResponseCommand(null);
-        final ConsumeMessageDirectlyResultRequestHeader requestHeader =
-                (ConsumeMessageDirectlyResultRequestHeader) request
-                    .decodeCommandCustomHeader(ConsumeMessageDirectlyResultRequestHeader.class);
-
         ClientChannelInfo clientChannelInfo =
-                this.brokerController.getConsumerManager().findChannel(requestHeader.getConsumerGroup(),
-                    requestHeader.getClientId());
+                this.brokerController.getConsumerManager().findChannel(consumerGroup, clientId);
 
         if (null == clientChannelInfo) {
             response.setCode(ResponseCode.SYSTEM_ERROR);
-            response.setRemark(String.format("The Consumer <%s> not online", requestHeader.getClientId()));
+            response.setRemark(String.format("The Consumer <%s> <%s> not online", consumerGroup, clientId));
             return response;
         }
 
         if (clientChannelInfo.getVersion() < MQVersion.Version.V3_1_8_SNAPSHOT.ordinal()) {
             response.setCode(ResponseCode.SYSTEM_ERROR);
             response.setRemark(String.format(
                 "The Consumer <%s> Version <%s> too low to finish, please upgrade it to V3_1_8_SNAPSHOT", //
-                requestHeader.getClientId(),//
+                clientId,//
                 MQVersion.getVersionDesc(clientChannelInfo.getVersion())));
             return response;
         }
 
         try {
-            RemotingCommand newRequest =
-                    RemotingCommand.createRequestCommand(RequestCode.CONSUME_MESSAGE_DIRECTLY, requestHeader);
+            RemotingCommand newRequest = RemotingCommand.createRequestCommand(requestCode, requestHeader);
             RemotingCommand consumerResponse =
-                    this.brokerController.getBroker2Client().getConsumerRunningInfo(
-                        clientChannelInfo.getChannel(), newRequest);
+                    this.brokerController.getBroker2Client().callClient(clientChannelInfo.getChannel(),
+                        newRequest);
             return consumerResponse;
         }
         catch (Exception e) {
             response.setCode(ResponseCode.SYSTEM_ERROR);
-            response.setRemark(String.format("invoke consumer <%s> Exception: %s",
-                requestHeader.getClientId(), RemotingHelper.exceptionSimpleDesc(e)));
+            response.setRemark(String.format("invoke consumer <%s> <%s> Exception: %s", consumerGroup,
+                clientId, RemotingHelper.exceptionSimpleDesc(e)));
             return response;
         }
     }
 
 
+    private RemotingCommand consumeMessageDirectly(ChannelHandlerContext ctx, RemotingCommand request)
+            throws RemotingCommandException {
+        final ConsumeMessageDirectlyResultRequestHeader requestHeader =
+                (ConsumeMessageDirectlyResultRequestHeader) request
+                    .decodeCommandCustomHeader(ConsumeMessageDirectlyResultRequestHeader.class);
+        return this.callConsumer(RequestCode.CONSUME_MESSAGE_DIRECTLY, requestHeader,
+            requestHeader.getConsumerGroup(), requestHeader.getClientId());
+    }
+
+
     /**
      * 调用Consumer，获取Consumer内存数据结构，为监控以及定位问题
      */
     private RemotingCommand getConsumerRunningInfo(ChannelHandlerContext ctx, RemotingCommand request)
             throws RemotingCommandException {
-        final RemotingCommand response = RemotingCommand.createResponseCommand(null);
         final GetConsumerRunningInfoRequestHeader requestHeader =
                 (GetConsumerRunningInfoRequestHeader) request
                     .decodeCommandCustomHeader(GetConsumerRunningInfoRequestHeader.class);
 
-        ClientChannelInfo clientChannelInfo =
-                this.brokerController.getConsumerManager().findChannel(requestHeader.getConsumerGroup(),
-                    requestHeader.getClientId());
-
-        if (null == clientChannelInfo) {
-            response.setCode(ResponseCode.SYSTEM_ERROR);
-            response.setRemark(String.format("The Consumer <%s> not online", requestHeader.getClientId()));
-            return response;
-        }
-
-        if (clientChannelInfo.getVersion() < MQVersion.Version.V3_1_8_SNAPSHOT.ordinal()) {
-            response.setCode(ResponseCode.SYSTEM_ERROR);
-            response.setRemark(String.format(
-                "The Consumer <%s> Version <%s> too low to finish, please upgrade it to V3_1_8_SNAPSHOT", //
-                requestHeader.getClientId(),//
-                MQVersion.getVersionDesc(clientChannelInfo.getVersion())));
-            return response;
-        }
-
-        try {
-            RemotingCommand newRequest =
-                    RemotingCommand
-                        .createRequestCommand(RequestCode.GET_CONSUMER_RUNNING_INFO, requestHeader);
-            RemotingCommand consumerResponse =
-                    this.brokerController.getBroker2Client().getConsumerRunningInfo(
-                        clientChannelInfo.getChannel(), newRequest);
-            return consumerResponse;
-        }
-        catch (Exception e) {
-            response.setCode(ResponseCode.SYSTEM_ERROR);
-            response.setRemark(String.format("invoke consumer <%s> Exception: %s",
-                requestHeader.getClientId(), RemotingHelper.exceptionSimpleDesc(e)));
-            return response;
-        }
+        return this.callConsumer(RequestCode.GET_CONSUMER_RUNNING_INFO, requestHeader,
+            requestHeader.getConsumerGroup(), requestHeader.getClientId());
     }