[ISSUE #9918] Fix the message may be renewed once more if the gRPC push consumer is unexpectedly disconnected (#9919)

Change-Id: I00424612c5ee9dd30e5c155dab17262051e5f097

Author: qianye1001

proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageActivity.java

@@ -135,6 +135,7 @@ public void receiveMessage(ProxyContext ctx, ReceiveMessageRequest request,
                     request.hasAttemptId() ? request.getAttemptId() : null,
                     timeRemaining
                 ).thenAccept(popResult -> {
+                    Runnable doAfterWrite = null;
                     if (proxyConfig.isEnableProxyAutoRenew() && request.getAutoRenew()) {
                         if (PopStatus.FOUND.equals(popResult.getPopStatus())) {
                             GrpcClientChannel clientChannel = grpcChannelManager.getChannel(ctx.getClientID());
@@ -145,19 +146,21 @@ public void receiveMessage(ProxyContext ctx, ReceiveMessageRequest request,
                                     writer.processThrowableWhenWriteMessage(e, ctx, request, messageExt));
                                 throw e;
                             }
-                            List<MessageExt> messageExtList = popResult.getMsgFoundList();
-                            for (MessageExt messageExt : messageExtList) {
-                                String receiptHandle = messageExt.getProperty(MessageConst.PROPERTY_POP_CK);
-                                if (receiptHandle != null) {
-                                    MessageReceiptHandle messageReceiptHandle =
-                                        new MessageReceiptHandle(group, topic, messageExt.getQueueId(), receiptHandle, messageExt.getMsgId(),
-                                            messageExt.getQueueOffset(), messageExt.getReconsumeTimes());
-                                    messagingProcessor.addReceiptHandle(ctx, clientChannel, group, messageExt.getMsgId(), messageReceiptHandle);
+                            doAfterWrite = () -> {
+                                List<MessageExt> messageExtList = popResult.getMsgFoundList();
+                                for (MessageExt messageExt : messageExtList) {
+                                    String receiptHandle = messageExt.getProperty(MessageConst.PROPERTY_POP_CK);
+                                    if (receiptHandle != null) {
+                                        MessageReceiptHandle messageReceiptHandle =
+                                            new MessageReceiptHandle(group, topic, messageExt.getQueueId(), receiptHandle, messageExt.getMsgId(),
+                                                messageExt.getQueueOffset(), messageExt.getReconsumeTimes());
+                                        messagingProcessor.addReceiptHandle(ctx, clientChannel, group, messageExt.getMsgId(), messageReceiptHandle);
+                                    }
                                 }
-                            }
+                            };
                         }
                     }
-                    writer.writeAndComplete(ctx, request, popResult);
+                    writer.writeAndComplete(ctx, request, popResult, doAfterWrite);
                 })
                 .exceptionally(t -> {
                     writer.writeAndComplete(ctx, request, t);

proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/consumer/ReceiveMessageResponseStreamWriter.java

@@ -54,6 +54,10 @@ public ReceiveMessageResponseStreamWriter(
     }
 
     public void writeAndComplete(ProxyContext ctx, ReceiveMessageRequest request, PopResult popResult) {
+        writeAndComplete(ctx, request, popResult, null);
+    }
+
+    public void writeAndComplete(ProxyContext ctx, ReceiveMessageRequest request, PopResult popResult, Runnable doAfterWrite) {
         PopStatus status = popResult.getPopStatus();
         List<MessageExt> messageFoundList = popResult.getMsgFoundList();
         try {
@@ -103,6 +107,9 @@ public void writeAndComplete(ProxyContext ctx, ReceiveMessageRequest request, Po
                         .build());
                     break;
             }
+            if (doAfterWrite != null) {
+                doAfterWrite.run();
+            }
         } catch (Throwable t) {
             writeResponseWithErrorIgnore(
                 ReceiveMessageResponse.newBuilder().setStatus(ResponseBuilder.getInstance().buildStatus(t)).build());