[ISSUE #816] Fix ack fail when Java Pushconsumer shutdown (#992)

qianye1001 · web-flow · commit 20757cc56037 · 2025-05-20T20:46:00.000+08:00
diff --git a/.github/workflows/csharp_build.yml b/.github/workflows/csharp_build.yml
@@ -8,7 +8,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        os: [ubuntu-20.04, macos-latest, windows-2022]
+        os: [ ubuntu-22.04, macos-latest, windows-2022 ]
     steps:
       - name: Checkout
         uses: actions/checkout@v3
diff --git a/.github/workflows/golang_build.yml b/.github/workflows/golang_build.yml
@@ -8,7 +8,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        os: [ubuntu-20.04, windows-2022]
+        os: [ ubuntu-22.04, windows-2022 ]
         go: [1.17]
     steps:
       - name: Checkout
diff --git a/.github/workflows/php_build.yml b/.github/workflows/php_build.yml
@@ -9,7 +9,7 @@ jobs:
       fail-fast: false
       matrix:
         php-version: ["7.4", "8.0", "8.1"]
-        os: [ubuntu-20.04, macos-11, windows-2022]
+        os: [ ubuntu-22.04, macos-11, windows-2022 ]
     steps:
       - name: Checkout
         uses: actions/checkout@v2
diff --git a/csharp/README.md b/csharp/README.md
@@ -38,7 +38,7 @@ dotnet add package RocketMQ.Client
 
 You can obtain the latest version of `RocketMQ.Client`
 from [NuGet Gallery](https://www.nuget.org/packages/RocketMQ.Client). To assist with getting started quickly and working
-with various message types and clients, we offer examples [here](./examples).
+with various message types and clients, we offer [examples](./examples).
 
 Layout of this project roughly
 follows [this guide](https://docs.microsoft.com/en-us/dotnet/core/tutorials/library-with-visual-studio-code?pivots=dotnet-5-0)
diff --git a/docs/design.md b/docs/design.md
@@ -66,7 +66,7 @@ Message identifier provides identity information for each message stored in brok
 
 >**Note**: Internal retries during message publishing may cause message duplication, the duplicate messages here have the same message identifier.
 
-The message identifier layout is redesigned, see more details [here](./message_id.md).
+The message identifier layout is redesigned, see more details in [message_id.md](./message_id.md).
 
 ## New and Unified APIs
 
diff --git a/java/README.md b/java/README.md
@@ -8,7 +8,8 @@ English | [简体中文](README-CN.md) | [RocketMQ Website](https://rocketmq.apa
 
 Here is the java implementation of the client for [Apache RocketMQ](https://rocketmq.apache.org/). Different from the [remoting-based client](https://github.com/apache/rocketmq/tree/develop/client), the current implementation is based on separating architecture for computing and storage, which is the more recommended way to access the RocketMQ service.
 
-Here are some preparations you may need to know (or refer to [here](https://rocketmq.apache.org/docs/quickStart/02quickstart/)).
+Here are some preparations you may need to know (or refer
+to [quick start](https://rocketmq.apache.org/docs/quickStart/02quickstart/)).
 
 1. Java 8+ for runtime, Java 11+ for the build;
 2. Setup namesrv, broker, and [proxy](https://github.com/apache/rocketmq/tree/develop/proxy).
@@ -57,7 +58,8 @@ implementation("org.apache.rocketmq:rocketmq-client-java-noshade:${rocketmq.vers
 implementation 'org.apache.rocketmq:rocketmq-client-java-noshade:${rocketmq.version}'
 ```
 
-More code examples are provided [here](./client/src/main/java/org/apache/rocketmq/client/java/example) to assist you in working with various clients and different message types.
+More code examples are provided [example](./client/src/main/java/org/apache/rocketmq/client/java/example) to assist you
+in working with various clients and different message types.
 
 ## Logging System
 
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/hook/CompositedMessageInterceptor.java b/java/client/src/main/java/org/apache/rocketmq/client/java/hook/CompositedMessageInterceptor.java
@@ -17,6 +17,7 @@
 
 package org.apache.rocketmq.client.java.hook;
 
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -29,10 +30,10 @@ public class CompositedMessageInterceptor implements MessageInterceptor {
     private static final Logger log = LoggerFactory.getLogger(MessageInterceptor.class);
     private static final AttributeKey<Map<Integer, Map<AttributeKey, Attribute>>> INTERCEPTOR_ATTRIBUTES_KEY =
         AttributeKey.create("composited_interceptor_attributes");
-    private final List<MessageInterceptor> interceptors;
+    private final List<MessageInterceptor> interceptors = new ArrayList<>();
 
     public CompositedMessageInterceptor(List<MessageInterceptor> interceptors) {
-        this.interceptors = interceptors;
+        this.interceptors.addAll(interceptors);
     }
 
     @Override
@@ -72,4 +73,8 @@ public void doAfter(MessageInterceptorContext context0, List<GeneralMessage> mes
             }
         }
     }
+    
+    public void addInterceptor(MessageInterceptor interceptor) {
+        interceptors.add(interceptor);
+    }
 }
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/hook/InflightRequestCountInterceptor.java b/java/client/src/main/java/org/apache/rocketmq/client/java/hook/InflightRequestCountInterceptor.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.client.java.hook;
+
+import java.util.List;
+import java.util.concurrent.atomic.AtomicLong;
+import org.apache.rocketmq.client.java.message.GeneralMessage;
+
+public class InflightRequestCountInterceptor implements MessageInterceptor {
+    private final AtomicLong inflightReceiveRequestCount = new AtomicLong(0);
+
+    @Override
+    public void doBefore(MessageInterceptorContext context, List<GeneralMessage> messages) {
+        if (context.getMessageHookPoints() == MessageHookPoints.RECEIVE) {
+            inflightReceiveRequestCount.incrementAndGet();
+        }
+    }
+
+    @Override
+    public void doAfter(MessageInterceptorContext context, List<GeneralMessage> messages) {
+        if (context.getMessageHookPoints() == MessageHookPoints.RECEIVE) {
+            inflightReceiveRequestCount.decrementAndGet();
+        }
+    }
+
+    public long getInflightReceiveRequestCount() {
+        return inflightReceiveRequestCount.get();
+    }
+}
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/ClientImpl.java
@@ -229,6 +229,12 @@ protected void shutDown() throws InterruptedException {
         log.info("Shutdown the rocketmq client successfully, clientId={}", clientId);
     }
 
+    protected void addMessageInterceptor(MessageInterceptor messageInterceptor) {
+        if (!this.isRunning()) {
+            compositedMessageInterceptor.addInterceptor(messageInterceptor);
+        }
+    }
+
     @Override
     public void doBefore(MessageInterceptorContext context, List<GeneralMessage> generalMessages) {
         try {
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ProcessQueueImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/ProcessQueueImpl.java
@@ -246,46 +246,46 @@ private void receiveMessageImmediately(String attemptId) {
             final ListenableFuture<ReceiveMessageResult> future = consumer.receiveMessage(request, mq,
                 longPollingTimeout);
             Futures.addCallback(future, new FutureCallback<ReceiveMessageResult>() {
-                @Override
-                public void onSuccess(ReceiveMessageResult result) {
-                    // Intercept after message reception.
-                    final List<GeneralMessage> generalMessages = result.getMessageViewImpls().stream()
-                        .map((Function<MessageView, GeneralMessage>) GeneralMessageImpl::new)
-                        .collect(Collectors.toList());
-                    final MessageInterceptorContextImpl context0 =
-                        new MessageInterceptorContextImpl(context, MessageHookPointsStatus.OK);
-                    consumer.doAfter(context0, generalMessages);
-
-                    try {
-                        onReceiveMessageResult(result);
-                    } catch (Throwable t) {
-                        // Should never reach here.
-                        log.error("[Bug] Exception raised while handling receive result, mq={}, endpoints={}, "
-                            + "clientId={}", mq, endpoints, clientId, t);
-                        onReceiveMessageException(t, attemptId);
+                    @Override
+                    public void onSuccess(ReceiveMessageResult result) {
+                        // Intercept after message reception.
+                        final List<GeneralMessage> generalMessages = result.getMessageViewImpls().stream()
+                            .map((Function<MessageView, GeneralMessage>) GeneralMessageImpl::new)
+                            .collect(Collectors.toList());
+                        final MessageInterceptorContextImpl context0 =
+                            new MessageInterceptorContextImpl(context, MessageHookPointsStatus.OK);
+                        consumer.doAfter(context0, generalMessages);
+
+                        try {
+                            onReceiveMessageResult(result);
+                        } catch (Throwable t) {
+                            // Should never reach here.
+                            log.error("[Bug] Exception raised while handling receive result, mq={}, endpoints={}, "
+                                + "clientId={}", mq, endpoints, clientId, t);
+                            onReceiveMessageException(t, attemptId);
+                        }
                     }
-                }
 
-                @Override
-                public void onFailure(Throwable t) {
-                    String nextAttemptId = null;
-                    if (t instanceof StatusRuntimeException) {
-                        StatusRuntimeException exception = (StatusRuntimeException) t;
-                        if (io.grpc.Status.DEADLINE_EXCEEDED.getCode() == exception.getStatus().getCode()) {
-                            nextAttemptId = request.getAttemptId();
+                    @Override
+                    public void onFailure(Throwable t) {
+                        String nextAttemptId = null;
+                        if (t instanceof StatusRuntimeException) {
+                            StatusRuntimeException exception = (StatusRuntimeException) t;
+                            if (io.grpc.Status.DEADLINE_EXCEEDED.getCode() == exception.getStatus().getCode()) {
+                                nextAttemptId = request.getAttemptId();
+                            }
                         }
+                        // Intercept after message reception.
+                        final MessageInterceptorContextImpl context0 =
+                            new MessageInterceptorContextImpl(context, MessageHookPointsStatus.ERROR);
+                        consumer.doAfter(context0, Collections.emptyList());
+
+                        log.error("Exception raised during message reception, mq={}, endpoints={}, attemptId={}, " +
+                                "nextAttemptId={}, clientId={}", mq, endpoints, request.getAttemptId(), nextAttemptId,
+                            clientId, t);
+                        onReceiveMessageException(t, nextAttemptId);
                     }
-                    // Intercept after message reception.
-                    final MessageInterceptorContextImpl context0 =
-                        new MessageInterceptorContextImpl(context, MessageHookPointsStatus.ERROR);
-                    consumer.doAfter(context0, Collections.emptyList());
-
-                    log.error("Exception raised during message reception, mq={}, endpoints={}, attemptId={}, " +
-                            "nextAttemptId={}, clientId={}", mq, endpoints, request.getAttemptId(), nextAttemptId,
-                        clientId, t);
-                    onReceiveMessageException(t, nextAttemptId);
-                }
-            }, MoreExecutors.directExecutor());
+                }, MoreExecutors.directExecutor());
             receptionTimes.getAndIncrement();
             consumer.getReceptionTimes().getAndIncrement();
         } catch (Throwable t) {
diff --git a/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/PushConsumerImpl.java b/java/client/src/main/java/org/apache/rocketmq/client/java/impl/consumer/PushConsumerImpl.java
@@ -58,6 +58,7 @@
 import org.apache.rocketmq.client.apis.consumer.PushConsumer;
 import org.apache.rocketmq.client.apis.message.MessageId;
 import org.apache.rocketmq.client.java.exception.StatusChecker;
+import org.apache.rocketmq.client.java.hook.InflightRequestCountInterceptor;
 import org.apache.rocketmq.client.java.hook.MessageHookPoints;
 import org.apache.rocketmq.client.java.hook.MessageHookPointsStatus;
 import org.apache.rocketmq.client.java.hook.MessageInterceptorContext;
@@ -103,6 +104,7 @@ class PushConsumerImpl extends ConsumerImpl implements PushConsumer {
     private final int maxCacheMessageCount;
     private final int maxCacheMessageSizeInBytes;
     private final boolean enableFifoConsumeAccelerator;
+    private final InflightRequestCountInterceptor inflightRequestCountInterceptor;
 
     /**
      * Indicates the times of message reception.
@@ -154,6 +156,9 @@ public PushConsumerImpl(ClientConfiguration clientConfiguration, String consumer
             TimeUnit.SECONDS,
             new LinkedBlockingQueue<>(),
             new ThreadFactoryImpl("MessageConsumption", this.getClientId().getIndex()));
+
+        this.inflightRequestCountInterceptor = new InflightRequestCountInterceptor();
+        this.addMessageInterceptor(inflightRequestCountInterceptor);
     }
 
     public PushConsumerImpl(ClientConfiguration clientConfiguration, String consumerGroup,
@@ -188,18 +193,54 @@ protected void startUp() throws Exception {
         }
     }
 
+    /**
+     * PushConsumerImpl shutdown order
+     * 1. when begin shutdown, do not send any new receive request
+     * 2. cancel scanAssignmentsFuture, do not create new processQueue
+     * 3. waiting all inflight receive request finished or timeout
+     * 4. shutdown consumptionExecutor and waiting all message consumption finished
+     * 5. sleep 1s to ack message async
+     * 6. shutdown clientImpl
+     */
     @Override
     protected void shutDown() throws InterruptedException {
         log.info("Begin to shutdown the rocketmq push consumer, clientId={}", clientId);
         if (null != scanAssignmentsFuture) {
             scanAssignmentsFuture.cancel(false);
         }
-        super.shutDown();
+        log.info("Waiting for the inflight receive requests to be finished, clientId={}", clientId);
+        waitingReceiveRequestFinished();
+        log.info("Begin to Shutdown consumption executor, clientId={}", clientId);
         this.consumptionExecutor.shutdown();
         ExecutorServices.awaitTerminated(consumptionExecutor);
+        TimeUnit.SECONDS.sleep(1);
+        super.shutDown();
         log.info("Shutdown the rocketmq push consumer successfully, clientId={}", clientId);
     }
 
+    private void waitingReceiveRequestFinished() {
+        Duration maxWaitingTime = clientConfiguration.getRequestTimeout()
+            .plus(pushSubscriptionSettings.getLongPollingTimeout());
+        long endTime = System.currentTimeMillis() + maxWaitingTime.toMillis();
+        try {
+            while (true) {
+                long inflightReceiveRequestCount = inflightRequestCountInterceptor.getInflightReceiveRequestCount();
+                if (inflightReceiveRequestCount <= 0) {
+                    log.info("All inflight receive requests have been finished, clientId={}", clientId);
+                    break;
+                } else if (System.currentTimeMillis() > endTime) {
+                    log.warn("Timeout waiting for all inflight receive requests to be finished, clientId={}, "
+                        + "inflightReceiveRequestCount={}", clientId, inflightReceiveRequestCount);
+                    break;
+                }
+                TimeUnit.MILLISECONDS.sleep(100);
+            }
+        } catch (Exception e) {
+            log.error("Unexpected exception while waiting for the inflight receive requests to be finished, "
+                + "clientId={}", clientId, e);
+        }
+    }
+
     private ConsumeService createConsumeService() {
         final ScheduledExecutorService scheduler = this.getClientManager().getScheduler();
         if (pushSubscriptionSettings.isFifo()) {
diff --git a/nodejs/README.md b/nodejs/README.md
@@ -4,7 +4,8 @@ English | [简体中文](README-CN.md) | [RocketMQ Website](https://rocketmq.apa
 
 ## Overview
 
-Here are some preparations you may need to know (or refer to [here](https://rocketmq.apache.org/docs/quickStart/01quickstart/)).
+Here are some preparations you may need to know (or refer
+to [quick start](https://rocketmq.apache.org/docs/quickStart/01quickstart/)).
 
 1. [Node.js](https://nodejs.dev/en/download/) 16.19.0 is the minimum version required, Node.js >= 18.17.0 is the recommended version.
 2. Setup namesrv, broker, and [proxy](https://github.com/apache/rocketmq/tree/develop/proxy).

Original file line number	Diff line number	Diff line change
`@@ -17,6 +17,7 @@`
`17`	`17`
`18`	`18`	`package org.apache.rocketmq.client.java.hook;`
`19`	`19`
	`20`	`+import java.util.ArrayList;`
`20`	`21`	`import java.util.HashMap;`
`21`	`22`	`import java.util.List;`
`22`	`23`	`import java.util.Map;`
`@@ -29,10 +30,10 @@ public class CompositedMessageInterceptor implements MessageInterceptor {`
`29`	`30`	`private static final Logger log = LoggerFactory.getLogger(MessageInterceptor.class);`
`30`	`31`	`private static final AttributeKey<Map<Integer, Map<AttributeKey, Attribute>>> INTERCEPTOR_ATTRIBUTES_KEY =`
`31`	`32`	`AttributeKey.create("composited_interceptor_attributes");`
`32`		`- private final List<MessageInterceptor> interceptors;`
	`33`	`+ private final List<MessageInterceptor> interceptors = new ArrayList<>();`
`33`	`34`
`34`	`35`	`public CompositedMessageInterceptor(List<MessageInterceptor> interceptors) {`
`35`		`- this.interceptors = interceptors;`
	`36`	`+ this.interceptors.addAll(interceptors);`
`36`	`37`	`}`
`37`	`38`
`38`	`39`	`@Override`
`@@ -72,4 +73,8 @@ public void doAfter(MessageInterceptorContext context0, List<GeneralMessage> mes`
`72`	`73`	`}`
`73`	`74`	`}`
`74`	`75`	`}`
	`76`	`+`
	`77`	`+ public void addInterceptor(MessageInterceptor interceptor) {`
	`78`	`+ interceptors.add(interceptor);`
	`79`	`+ }`
`75`	`80`	`}`