[ISSUE #10774] Reject empty producer message lists

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/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);