From 27604594abaa030e263563f4d12c743207789166 Mon Sep 17 00:00:00 2001 From: liuhy Date: Mon, 3 Aug 2026 00:07:53 -0700 Subject: [PATCH 1/2] [ISSUE #10780] Reject empty cluster batch ack handles --- .../message/ClusterMessageService.java | 6 +++ .../message/ClusterMessageServiceTest.java | 39 +++++++++++++++++-- 2 files changed, 42 insertions(+), 3 deletions(-) diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/ClusterMessageService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/ClusterMessageService.java index 77c4ef60f14..b4984c9edd6 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/ClusterMessageService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/ClusterMessageService.java @@ -159,6 +159,12 @@ public CompletableFuture ackMessage(ProxyContext ctx, ReceiptHandle h public CompletableFuture batchAckMessage(ProxyContext ctx, List handleList, String consumerGroup, String topic, long timeoutMillis) { + if (handleList == null || handleList.isEmpty()) { + return FutureUtils.completeExceptionally(new ProxyException( + ProxyExceptionCode.INVALID_RECEIPT_HANDLE, + "receipt handle list is empty" + )); + } List extraInfoList = handleList.stream().map(message -> message.getReceiptHandle().getReceiptHandle()).collect(Collectors.toList()); return this.mqClientAPIFactory.getClient().batchAckMessageAsync( this.resolveBrokerAddrInReceiptHandle(ctx, handleList.get(0).getReceiptHandle()), diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/ClusterMessageServiceTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/ClusterMessageServiceTest.java index 7e4d25f0c09..0ebac017f3a 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/ClusterMessageServiceTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/ClusterMessageServiceTest.java @@ -16,13 +16,15 @@ */ package org.apache.rocketmq.proxy.service.message; +import java.util.Collections; +import java.util.concurrent.ExecutionException; import org.apache.rocketmq.client.exception.MQClientException; +import org.apache.rocketmq.client.impl.mqclient.MQClientAPIFactory; import org.apache.rocketmq.common.consumer.ReceiptHandle; import org.apache.rocketmq.common.message.MessageClientIDSetter; -import org.apache.rocketmq.proxy.common.ProxyContext; import org.apache.rocketmq.proxy.common.ProxyException; import org.apache.rocketmq.proxy.common.ProxyExceptionCode; -import org.apache.rocketmq.client.impl.mqclient.MQClientAPIFactory; +import org.apache.rocketmq.proxy.common.ProxyContext; import org.apache.rocketmq.proxy.service.route.TopicRouteService; import org.apache.rocketmq.remoting.protocol.ResponseCode; import org.apache.rocketmq.remoting.protocol.header.AckMessageRequestHeader; @@ -35,17 +37,20 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; public class ClusterMessageServiceTest { private TopicRouteService topicRouteService; + private MQClientAPIFactory mqClientAPIFactory; private ClusterMessageService clusterMessageService; @Before public void before() { this.topicRouteService = mock(TopicRouteService.class); - MQClientAPIFactory mqClientAPIFactory = mock(MQClientAPIFactory.class); + this.mqClientAPIFactory = mock(MQClientAPIFactory.class); this.clusterMessageService = new ClusterMessageService(this.topicRouteService, mqClientAPIFactory); } @@ -76,4 +81,32 @@ public void testAckMessageByInvalidBrokerNameHandle() throws Exception { assertEquals(ProxyExceptionCode.INVALID_RECEIPT_HANDLE, proxyException.getCode()); } } + + @Test + public void testBatchAckMessageByEmptyHandleList() throws Exception { + assertInvalidBatchAckHandleList(Collections.emptyList()); + } + + @Test + public void testBatchAckMessageByNullHandleList() throws Exception { + assertInvalidBatchAckHandleList(null); + } + + private void assertInvalidBatchAckHandleList(java.util.List handleList) throws Exception { + try { + this.clusterMessageService.batchAckMessage( + ProxyContext.create(), + handleList, + "consumerGroup", + "topic", + 3000 + ).get(); + fail(); + } catch (ExecutionException e) { + assertTrue(e.getCause() instanceof ProxyException); + ProxyException proxyException = (ProxyException) e.getCause(); + assertEquals(ProxyExceptionCode.INVALID_RECEIPT_HANDLE, proxyException.getCode()); + } + verify(this.mqClientAPIFactory, never()).getClient(); + } } From 8e0e6ebe2e597de3a850cbc7e97d4959fc05b7be Mon Sep 17 00:00:00 2001 From: liuhy Date: Tue, 4 Aug 2026 04:01:02 -0700 Subject: [PATCH 2/2] fix(proxy): clarify invalid batch ack handles --- .../rocketmq/proxy/service/message/ClusterMessageService.java | 2 +- .../proxy/service/message/ClusterMessageServiceTest.java | 1 + 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/ClusterMessageService.java b/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/ClusterMessageService.java index b4984c9edd6..3f50e70e942 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/ClusterMessageService.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/service/message/ClusterMessageService.java @@ -162,7 +162,7 @@ public CompletableFuture batchAckMessage(ProxyContext ctx, List extraInfoList = handleList.stream().map(message -> message.getReceiptHandle().getReceiptHandle()).collect(Collectors.toList()); diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/ClusterMessageServiceTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/ClusterMessageServiceTest.java index 0ebac017f3a..2b8c7afa474 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/ClusterMessageServiceTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/service/message/ClusterMessageServiceTest.java @@ -106,6 +106,7 @@ private void assertInvalidBatchAckHandleList(java.util.List