指定Consumer消费某条消息。增加日志。

Author: vintagewang

rocketmq-client/src/main/java/com/alibaba/rocketmq/client/impl/consumer/ConsumeMessageConcurrentlyService.java

@@ -428,10 +428,18 @@ public ConsumeMessageDirectlyResult consumeMessageDirectly(MessageExt msg, Strin
         catch (Throwable e) {
             result.setConsumeResult(CMResult.CR_THROW_EXCEPTION);
             result.setRemark(RemotingHelper.exceptionSimpleDesc(e));
+
+            log.warn(String.format("consumeMessageDirectly exception: {} Group: {} Msgs: {} MQ: {}",//
+                RemotingHelper.exceptionSimpleDesc(e),//
+                ConsumeMessageConcurrentlyService.this.consumerGroup,//
+                msgs,//
+                mq), e);
         }
 
         result.setSpentTimeMills(System.currentTimeMillis() - beginTime);
 
+        log.warn("consumeMessageDirectly Result: {}", result);
+
         return result;
     }
 

rocketmq-client/src/main/java/com/alibaba/rocketmq/client/impl/consumer/ConsumeMessageOrderlyService.java

@@ -529,11 +529,19 @@ public ConsumeMessageDirectlyResult consumeMessageDirectly(MessageExt msg, Strin
         catch (Throwable e) {
             result.setConsumeResult(CMResult.CR_THROW_EXCEPTION);
             result.setRemark(RemotingHelper.exceptionSimpleDesc(e));
+
+            log.warn(String.format("consumeMessageDirectly exception: {} Group: {} Msgs: {} MQ: {}",//
+                RemotingHelper.exceptionSimpleDesc(e),//
+                ConsumeMessageOrderlyService.this.consumerGroup,//
+                msgs,//
+                mq), e);
         }
 
         result.setAutoCommit(context.isAutoCommit());
         result.setSpentTimeMills(System.currentTimeMillis() - beginTime);
 
+        log.warn("consumeMessageDirectly Result: {}", result);
+
         return result;
     }