Set attemptId for ReceiveMessageRequest

apache · Jun 14, 2023 · ad2279c · ad2279c
1 parent edcd7f9
commit ad2279c
Showing 1 changed file with 9 additions and 4 deletions.
diff --git a/.../client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ProcessQueueImpl.java b/.../client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ProcessQueueImpl.java
@@ -37,6 +37,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.UUID;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
@@ -198,8 +199,12 @@ private void receiveMessageLater(Duration delay, String attemptId) {
         }
     }
 
+    private String generateAttemptId() {
+        return UUID.randomUUID().toString();
+    }
+
     public void receiveMessage() {
-        receiveMessage(null);
+        receiveMessage(this.generateAttemptId());
     }
 
     public void receiveMessage(String attemptId) {
@@ -217,7 +222,7 @@ public void receiveMessage(String attemptId) {
     }
 
     private void receiveMessageImmediately() {
-        receiveMessageImmediately(null);
+        receiveMessageImmediately(this.generateAttemptId());
     }
 
     private void receiveMessageImmediately(String attemptId) {
@@ -275,8 +280,8 @@ public void onFailure(Throwable t) {
                         new MessageInterceptorContextImpl(context, MessageHookPointsStatus.ERROR);
                     consumer.doAfter(context0, Collections.emptyList());
 
-                    log.error("Exception raised during message reception, mq={}, endpoints={}, clientId={}", mq,
-                        endpoints, clientId, t);
+                    log.error("Exception raised during message reception, mq={}, endpoints={}, attemptId={}," +
+                            " clientId={}", mq, endpoints, request.getAttemptId(), clientId, t);
                     onReceiveMessageException(t, attemptId);
                 }
             }, MoreExecutors.directExecutor());