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..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 @@ -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 null or 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..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 @@ -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,33 @@ 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()); + assertEquals("receipt handle list is null or empty", proxyException.getMessage()); + } + verify(this.mqClientAPIFactory, never()).getClient(); + } }