[ISSUE #10933] Retain receipt handles when offline cleanup cannot complete

proxy/src/main/java/org/apache/rocketmq/proxy/service/receipt/DefaultReceiptHandleManager.java

@@ -24,6 +24,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
@@ -89,7 +90,8 @@ public DefaultReceiptHandleManager(MetadataService metadataService, ConsumerMana
             proxyConfig.getReturnHandleGroupThreadPoolNums() * 2,
             1, TimeUnit.MINUTES,
             "ReturnHandleGroupWorkerThread",
-            proxyConfig.getRenewThreadPoolQueueCapacity()
+            proxyConfig.getRenewThreadPoolQueueCapacity(),
+            new ThreadPoolExecutor.AbortPolicy()
         );
         consumerManager.appendConsumerIdsChangeListener(new ConsumerIdsChangeListener() {
             @Override
@@ -251,7 +253,14 @@ protected void clearGroup(ReceiptHandleGroupKey key) {
             return;
         }
         ReceiptHandleGroup handleGroup = receiptHandleGroupMap.remove(key);
-        returnHandleGroupWorkerService.submit(() -> returnHandleGroup(key, handleGroup));
+        try {
+            returnHandleGroupWorkerService.submit(() -> returnHandleGroup(key, handleGroup));
+        } catch (RejectedExecutionException e) {
+            if (handleGroup != null) {
+                receiptHandleGroupMap.putIfAbsent(key, handleGroup);
+            }
+            log.warn("submit clear handle group task failed, will retry in the next schedule. key:{}", key, e);
+        }
     }
 
     // There is no longer any waiting for lock, and only the locked handles will be processed immediately,
@@ -266,7 +275,7 @@ private void returnHandleGroup(ReceiptHandleGroupKey key, ReceiptHandleGroup han
                 handleGroup.computeIfPresent(msgID, handle, messageReceiptHandle -> {
                     CompletableFuture<AckResult> future = new CompletableFuture<>();
                     eventListener.fireEvent(new RenewEvent(key, messageReceiptHandle, proxyConfig.getInvisibleTimeMillisWhenClear(), RenewEvent.EventType.CLEAR_GROUP, future));
-                    return CompletableFuture.completedFuture(null);
+                    return future.thenApply(ackResult -> null);
                 }, 0);
             } catch (Exception e) {
                 log.error("error when clear handle for group. key:{}", key, e);

proxy/src/test/java/org/apache/rocketmq/proxy/service/receipt/DefaultReceiptHandleManagerTest.java

@@ -60,6 +60,7 @@
 
 import static org.awaitility.Awaitility.await;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 
 public class DefaultReceiptHandleManagerTest extends BaseServiceTest {
@@ -454,6 +455,37 @@ public void testClearGroup() {
                 Mockito.eq(GROUP), Mockito.eq(TOPIC), Mockito.eq(ConfigurationManager.getProxyConfig().getInvisibleTimeMillisWhenClear()));
     }
 
+    @Test
+    public void testClearGroupRetainsHandleWhenRenewalFails() {
+        Channel channel = PROXY_CONTEXT.getVal(ContextVariable.CHANNEL);
+        ReceiptHandleGroupKey key = new ReceiptHandleGroupKey(channel, GROUP);
+        receiptHandleManager.addReceiptHandle(PROXY_CONTEXT, channel, GROUP, MSG_ID, messageReceiptHandle);
+        CompletableFuture<AckResult> failedFuture = new CompletableFuture<>();
+        failedFuture.completeExceptionally(new MQClientException(0, "renew failed"));
+        Mockito.when(messagingProcessor.changeInvisibleTime(Mockito.any(ProxyContext.class), Mockito.any(ReceiptHandle.class),
+                Mockito.eq(MESSAGE_ID), Mockito.eq(GROUP), Mockito.eq(TOPIC),
+                Mockito.eq(ConfigurationManager.getProxyConfig().getInvisibleTimeMillisWhenClear())))
+            .thenReturn(failedFuture);
+
+        receiptHandleManager.clearGroup(key);
+
+        await().atMost(Duration.ofSeconds(1)).until(() ->
+            receiptHandleManager.receiptHandleGroupMap.containsKey(key));
+        assertFalse(receiptHandleManager.receiptHandleGroupMap.get(key).isEmpty());
+    }
+
+    @Test
+    public void testClearGroupRetainsHandlesWhenCleanupSubmissionIsRejected() {
+        Channel channel = PROXY_CONTEXT.getVal(ContextVariable.CHANNEL);
+        ReceiptHandleGroupKey key = new ReceiptHandleGroupKey(channel, GROUP);
+        receiptHandleManager.addReceiptHandle(PROXY_CONTEXT, channel, GROUP, MSG_ID, messageReceiptHandle);
+        receiptHandleManager.returnHandleGroupWorkerService.shutdownNow();
+
+        receiptHandleManager.clearGroup(key);
+
+        assertTrue(receiptHandleManager.receiptHandleGroupMap.containsKey(key));
+    }
+
     @Test
     public void testClientOffline() {
         ArgumentCaptor<ConsumerIdsChangeListener> listenerArgumentCaptor = ArgumentCaptor.forClass(ConsumerIdsChangeListener.class);
@@ -463,4 +495,4 @@ public void testClientOffline() {
         listenerArgumentCaptor.getValue().handle(ConsumerGroupEvent.CLIENT_UNREGISTER, GROUP, new ClientChannelInfo(channel, "", LanguageCode.JAVA, 0));
         assertTrue(receiptHandleManager.receiptHandleGroupMap.isEmpty());
     }
-}
+}