fix: validate empty proxy producer, ack, and lock requests

proxy/src/main/java/org/apache/rocketmq/proxy/common/ProxyExceptionCode.java

@@ -18,6 +18,7 @@
 
 public enum ProxyExceptionCode {
     INVALID_BROKER_NAME,
+    INVALID_REQUEST,
     TRANSACTION_DATA_NOT_FOUND,
     FORBIDDEN,
     MESSAGE_PROPERTY_CONFLICT_WITH_TYPE,

proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcProxyException.java

@@ -31,6 +31,7 @@ public class GrpcProxyException extends RuntimeException {
 
     static {
         CODE_MAPPING.put(ProxyExceptionCode.INVALID_BROKER_NAME, Code.BAD_REQUEST);
+        CODE_MAPPING.put(ProxyExceptionCode.INVALID_REQUEST, Code.BAD_REQUEST);
         CODE_MAPPING.put(ProxyExceptionCode.INVALID_RECEIPT_HANDLE, Code.INVALID_RECEIPT_HANDLE);
         CODE_MAPPING.put(ProxyExceptionCode.FORBIDDEN, Code.FORBIDDEN);
         CODE_MAPPING.put(ProxyExceptionCode.INTERNAL_SERVER_ERROR, Code.INTERNAL_SERVER_ERROR);

proxy/src/main/java/org/apache/rocketmq/proxy/processor/ProducerProcessor.java

@@ -72,6 +72,9 @@ public CompletableFuture<List<SendResult>> sendMessage(ProxyContext ctx, QueueSe
         long beginTimestampFirst = System.currentTimeMillis();
         AddressableMessageQueue messageQueue = null;
         try {
+            if (messageList == null || messageList.isEmpty()) {
+                throw new ProxyException(ProxyExceptionCode.INVALID_REQUEST, "message list is empty");
+            }
             Message message = messageList.get(0);
             String topic = message.getTopic();
             if (isNeedCheckTopicMessageType(message)) {

proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/AbstractRemotingActivity.java

@@ -49,6 +49,7 @@ public abstract class AbstractRemotingActivity implements NettyRequestProcessor
     private static final Map<ProxyExceptionCode, Integer> PROXY_EXCEPTION_RESPONSE_CODE_MAP = new HashMap<ProxyExceptionCode, Integer>() {
         {
             put(ProxyExceptionCode.FORBIDDEN, ResponseCode.NO_PERMISSION);
+            put(ProxyExceptionCode.INVALID_REQUEST, ResponseCode.MESSAGE_ILLEGAL);
             put(ProxyExceptionCode.MESSAGE_PROPERTY_CONFLICT_WITH_TYPE, ResponseCode.MESSAGE_ILLEGAL);
             put(ProxyExceptionCode.INTERNAL_SERVER_ERROR, ResponseCode.SYSTEM_ERROR);
             put(ProxyExceptionCode.TRANSACTION_DATA_NOT_FOUND, ResponseCode.SUCCESS);

proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/ConsumerManagerActivity.java

@@ -45,6 +45,8 @@
 import org.apache.rocketmq.remoting.protocol.RemotingCommand;
 
 public class ConsumerManagerActivity extends AbstractRemotingActivity {
+    static final String EMPTY_QUEUE_REMARK = "MessageQueue set is empty";
+
     public ConsumerManagerActivity(RequestPipeline requestPipeline, MessagingProcessor messagingProcessor) {
         super(requestPipeline, messagingProcessor);
     }
@@ -136,7 +138,8 @@ protected RemotingCommand lockBatchMQ(ChannelHandlerContext ctx, RemotingCommand
         Set<MessageQueue> mqSet = requestBody.getMqSet();
         if (mqSet.isEmpty()) {
             response.setBody(requestBody.encode());
-            response.setRemark("MessageQueue set is empty");
+            response.setCode(ResponseCode.INVALID_PARAMETER);
+            response.setRemark(EMPTY_QUEUE_REMARK);
             return response;
         }
 
@@ -157,7 +160,8 @@ protected RemotingCommand unlockBatchMQ(ChannelHandlerContext ctx, RemotingComma
         Set<MessageQueue> mqSet = requestBody.getMqSet();
         if (mqSet.isEmpty()) {
             response.setBody(requestBody.encode());
-            response.setRemark("MessageQueue set is empty");
+            response.setCode(ResponseCode.INVALID_PARAMETER);
+            response.setRemark(EMPTY_QUEUE_REMARK);
             return response;
         }
 

proxy/src/main/java/org/apache/rocketmq/proxy/service/message/ClusterMessageService.java

@@ -159,6 +159,12 @@ public CompletableFuture<AckResult> ackMessage(ProxyContext ctx, ReceiptHandle h
     public CompletableFuture<AckResult> batchAckMessage(ProxyContext ctx, List<ReceiptHandleMessage> 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<String> extraInfoList = handleList.stream().map(message -> message.getReceiptHandle().getReceiptHandle()).collect(Collectors.toList());
         return this.mqClientAPIFactory.getClient().batchAckMessageAsync(
             this.resolveBrokerAddrInReceiptHandle(ctx, handleList.get(0).getReceiptHandle()),

proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcProxyExceptionTest.java

@@ -0,0 +1,36 @@
+/*
+ * 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.proxy.grpc.v2.common;
+
+import apache.rocketmq.v2.Code;
+import org.apache.rocketmq.proxy.common.ProxyException;
+import org.apache.rocketmq.proxy.common.ProxyExceptionCode;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+
+public class GrpcProxyExceptionTest {
+
+    @Test
+    public void testInvalidRequestMapsToBadRequest() {
+        GrpcProxyException exception = new GrpcProxyException(
+            new ProxyException(ProxyExceptionCode.INVALID_REQUEST, "message list is empty"));
+
+        assertEquals(Code.BAD_REQUEST, exception.getCode());
+    }
+}

proxy/src/test/java/org/apache/rocketmq/proxy/processor/ProducerProcessorTest.java

@@ -19,6 +19,7 @@
 
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
@@ -75,6 +76,44 @@ public void before() throws Throwable {
         this.producerProcessor = new ProducerProcessor(this.messagingProcessor, this.serviceManager, Executors.newCachedThreadPool());
     }
 
+    @Test
+    public void testSendMessageRejectsEmptyMessageList() {
+        CompletionException exception = Assert.assertThrows(CompletionException.class, () -> {
+            this.producerProcessor.sendMessage(
+                createContext(),
+                (ctx, messageQueueView) -> null,
+                PRODUCER_GROUP,
+                MessageSysFlag.TRANSACTION_NOT_TYPE,
+                Collections.emptyList(),
+                3000
+            ).join();
+        });
+
+        assertTrue(exception.getCause() instanceof ProxyException);
+        ProxyException cause = (ProxyException) exception.getCause();
+        assertEquals(ProxyExceptionCode.INVALID_REQUEST, cause.getCode());
+        assertEquals("message list is empty", cause.getMessage());
+    }
+
+    @Test
+    public void testSendMessageRejectsNullMessageList() {
+        CompletionException exception = Assert.assertThrows(CompletionException.class, () -> {
+            this.producerProcessor.sendMessage(
+                createContext(),
+                (ctx, messageQueueView) -> null,
+                PRODUCER_GROUP,
+                MessageSysFlag.TRANSACTION_NOT_TYPE,
+                null,
+                3000
+            ).join();
+        });
+
+        assertTrue(exception.getCause() instanceof ProxyException);
+        ProxyException cause = (ProxyException) exception.getCause();
+        assertEquals(ProxyExceptionCode.INVALID_REQUEST, cause.getCode());
+        assertEquals("message list is empty", cause.getMessage());
+    }
+
     @Test
     public void testSendMessage() throws Throwable {
         when(metadataService.getTopicMessageType(any(), eq(TOPIC))).thenReturn(TopicMessageType.NORMAL);

proxy/src/test/java/org/apache/rocketmq/proxy/remoting/activity/ConsumerManagerActivityTest.java

@@ -0,0 +1,78 @@
+/*
+ * 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.proxy.remoting.activity;
+
+import org.apache.rocketmq.proxy.config.InitConfigTest;
+import org.apache.rocketmq.proxy.processor.MessagingProcessor;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.apache.rocketmq.remoting.protocol.RequestCode;
+import org.apache.rocketmq.remoting.protocol.ResponseCode;
+import org.apache.rocketmq.remoting.protocol.body.LockBatchRequestBody;
+import org.apache.rocketmq.remoting.protocol.body.UnlockBatchRequestBody;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+
+@RunWith(MockitoJUnitRunner.class)
+public class ConsumerManagerActivityTest extends InitConfigTest {
+    ConsumerManagerActivity consumerManagerActivity;
+
+    @Mock
+    MessagingProcessor messagingProcessorMock;
+
+    @Before
+    public void setup() {
+        consumerManagerActivity = new ConsumerManagerActivity(null, messagingProcessorMock);
+    }
+
+    @Test
+    public void testLockBatchMQWithEmptyQueueSetReturnsErrorCode() throws Exception {
+        LockBatchRequestBody requestBody = new LockBatchRequestBody();
+        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.LOCK_BATCH_MQ, null);
+        request.setBody(requestBody.encode());
+
+        RemotingCommand response = consumerManagerActivity.lockBatchMQ(null, request, null);
+
+        assertThat(response.getCode()).isEqualTo(ResponseCode.INVALID_PARAMETER);
+        assertThat(response.getRemark()).isEqualTo(ConsumerManagerActivity.EMPTY_QUEUE_REMARK);
+        assertThat(response.getBody()).isEqualTo(requestBody.encode());
+        verify(messagingProcessorMock, never()).request(any(), any(), any(), anyLong());
+    }
+
+    @Test
+    public void testUnlockBatchMQWithEmptyQueueSetReturnsErrorCode() throws Exception {
+        UnlockBatchRequestBody requestBody = new UnlockBatchRequestBody();
+        RemotingCommand request = RemotingCommand.createRequestCommand(RequestCode.UNLOCK_BATCH_MQ, null);
+        request.setBody(requestBody.encode());
+
+        RemotingCommand response = consumerManagerActivity.unlockBatchMQ(null, request, null);
+
+        assertThat(response.getCode()).isEqualTo(ResponseCode.INVALID_PARAMETER);
+        assertThat(response.getRemark()).isEqualTo(ConsumerManagerActivity.EMPTY_QUEUE_REMARK);
+        assertThat(response.getBody()).isEqualTo(requestBody.encode());
+        verify(messagingProcessorMock, never()).request(any(), any(), any(), anyLong());
+    }
+}

proxy/src/test/java/org/apache/rocketmq/proxy/service/message/ClusterMessageServiceTest.java

@@ -76,4 +76,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<ReceiptHandleMessage> 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();
+    }
 }