add reconsume times header

spring-cloud-stream-binder-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/RocketMQBinderConstants.java

@@ -16,8 +16,11 @@
 
 package com.alibaba.cloud.stream.binder.rocketmq;
 
+import static org.apache.rocketmq.spring.support.RocketMQHeaders.PREFIX;
+
 /**
  * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
+ * @author <a href="mailto:jiashuai.xie01@gmail.com">Xiejiashuai</a>
  */
 public interface RocketMQBinderConstants {
 
@@ -31,6 +34,11 @@ public interface RocketMQBinderConstants {
      */
     String DEFAULT_NAME_SERVER = "127.0.0.1:9876";
 
-	String DEFAULT_GROUP = "rocketmq_binder_default_group_name";
+    String DEFAULT_GROUP = PREFIX + "binder_default_group_name";
+
+    /**
+     * RocketMQ re-consume times
+     */
+    String ROCKETMQ_RECONSUME_TIMES = PREFIX + "RECONSUME_TIMES";
 
 }

spring-cloud-stream-binder-rocketmq/src/main/java/com/alibaba/cloud/stream/binder/rocketmq/consuming/RocketMQListenerBindingContainer.java

@@ -16,19 +16,14 @@
 
 package com.alibaba.cloud.stream.binder.rocketmq.consuming;
 
-import java.util.List;
-import java.util.Objects;
-
+import com.alibaba.cloud.stream.binder.rocketmq.RocketMQMessageChannelBinder;
+import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQBinderConfigurationProperties;
+import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQConsumerProperties;
 import org.apache.rocketmq.acl.common.AclClientRPCHook;
 import org.apache.rocketmq.acl.common.SessionCredentials;
 import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
 import org.apache.rocketmq.client.consumer.MessageSelector;
-import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
-import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
-import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
-import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
-import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
-import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
+import org.apache.rocketmq.client.consumer.listener.*;
 import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
 import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.common.UtilAll;
@@ -49,12 +44,19 @@ import org.springframework.context.SmartLifecycle;
 import org.springframework.util.Assert;
 import org.springframework.util.StringUtils;
 
-import com.alibaba.cloud.stream.binder.rocketmq.RocketMQMessageChannelBinder;
-import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQBinderConfigurationProperties;
-import com.alibaba.cloud.stream.binder.rocketmq.properties.RocketMQConsumerProperties;
+import java.util.List;
+import java.util.Objects;
+
+import static com.alibaba.cloud.stream.binder.rocketmq.RocketMQBinderConstants.ROCKETMQ_RECONSUME_TIMES;
 
 /**
+ * A class that Listen on rocketmq message
+ * <p>
+ * this class will delegate  {@link RocketMQListener} to handle message
+ *
  * @author <a href="mailto:fangjian0423@gmail.com">Jim</a>
+ * @author <a href="mailto:jiashuai.xie01@gmail.com">Xiejiashuai</a>
+ * @see RocketMQListener
  */
 public class RocketMQListenerBindingContainer
         implements InitializingBean, RocketMQListenerContainer, SmartLifecycle {
@@ -112,8 +114,7 @@ public class RocketMQListenerBindingContainer
         if (StringUtils.isEmpty(rocketMQConsumerProperties.getExtension().getSql())) {
             this.selectorType = SelectorType.TAG;
             this.selectorExpression = rocketMQConsumerProperties.getExtension().getTags();
-		}
-		else {
+        } else {
             this.selectorType = SelectorType.SQL92;
             this.selectorExpression = rocketMQConsumerProperties.getExtension().getSql();
         }
@@ -161,8 +162,7 @@ public class RocketMQListenerBindingContainer
 
         try {
             consumer.start();
-		}
-		catch (MQClientException e) {
+        } catch (MQClientException e) {
             throw new IllegalStateException("Failed to start RocketMQ push consumer", e);
         }
         this.setRunning(true);
@@ -211,8 +211,7 @@ public class RocketMQListenerBindingContainer
             consumer.setInstanceName(RocketMQUtil.getInstanceName(rpcHook,
                     topic + "|" + UtilAll.getPid()));
             consumer.setVipChannelEnabled(false);
-		}
-		else {
+        } else {
             consumer = new DefaultMQPushConsumer(consumerGroup,
                     rocketBinderConfigurationProperties.isEnableMsgTrace(),
                     rocketBinderConfigurationProperties.getCustomizedTraceTopic());
@@ -368,7 +367,7 @@ public class RocketMQListenerBindingContainer
     public class DefaultMessageListenerConcurrently
             implements MessageListenerConcurrently {
 
-		@SuppressWarnings("unchecked")
+        @SuppressWarnings({"unchecked", "Duplicates"})
         @Override
         public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                         ConsumeConcurrentlyContext context) {
@@ -376,12 +375,12 @@ public class RocketMQListenerBindingContainer
                 log.debug("received msg: {}", messageExt);
                 try {
                     long now = System.currentTimeMillis();
+                    preOnMessage(messageExt);
                     rocketMQListener
                             .onMessage(RocketMQUtil.convertToSpringMessage(messageExt));
                     long costTime = System.currentTimeMillis() - now;
-					log.debug("consume {} cost: {} ms", messageExt.getMsgId(), costTime);
-				}
-				catch (Exception e) {
+                    log.debug("consume {} message key:[{}] cost: {} ms", messageExt.getMsgId(), messageExt.getKeys(), costTime);
+                } catch (Exception e) {
                     log.warn("consume message failed. messageExt:{}", messageExt, e);
                     context.setDelayLevelWhenNextConsume(delayLevelWhenNextConsume);
                     return ConsumeConcurrentlyStatus.RECONSUME_LATER;
@@ -394,7 +393,7 @@ public class RocketMQListenerBindingContainer
 
     public class DefaultMessageListenerOrderly implements MessageListenerOrderly {
 
-		@SuppressWarnings("unchecked")
+        @SuppressWarnings({"unchecked", "Duplicates"})
         @Override
         public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs,
                                                    ConsumeOrderlyContext context) {
@@ -402,12 +401,12 @@ public class RocketMQListenerBindingContainer
                 log.debug("received msg: {}", messageExt);
                 try {
                     long now = System.currentTimeMillis();
+                    preOnMessage(messageExt);
                     rocketMQListener
                             .onMessage(RocketMQUtil.convertToSpringMessage(messageExt));
                     long costTime = System.currentTimeMillis() - now;
-					log.info("consume {} cost: {} ms", messageExt.getMsgId(), costTime);
-				}
-				catch (Exception e) {
+                    log.debug("consume {} message key:[{}] cost: {} ms", messageExt.getMsgId(), messageExt.getKeys(), costTime);
+                } catch (Exception e) {
                     log.warn("consume message failed. messageExt:{}", messageExt, e);
                     context.setSuspendCurrentQueueTimeMillis(
                             suspendCurrentQueueTimeMillis);
@@ -419,4 +418,14 @@ public class RocketMQListenerBindingContainer
         }
     }
 
+    /**
+     * pre handle message before the consumer handle message
+     *
+     * @param messageExt the rocketmq message
+     */
+    private void preOnMessage(MessageExt messageExt) {
+        int reconsumeTimes = messageExt.getReconsumeTimes();
+        messageExt.putUserProperty(ROCKETMQ_RECONSUME_TIMES, String.valueOf(reconsumeTimes));
+    }
+
 }