From 37d8991bb0f756b3406c3e35f7d8e2cee487214e Mon Sep 17 00:00:00 2001 From: liuhy Date: Sun, 2 Aug 2026 23:46:53 -0700 Subject: [PATCH 1/2] [ISSUE #10774] Reject empty producer message lists --- .../proxy/processor/ProducerProcessor.java | 3 ++ .../processor/ProducerProcessorTest.java | 39 +++++++++++++++++++ 2 files changed, 42 insertions(+) diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ProducerProcessor.java b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ProducerProcessor.java index 8c4907c588a..881f0b8593f 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ProducerProcessor.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ProducerProcessor.java @@ -72,6 +72,9 @@ public CompletableFuture> sendMessage(ProxyContext ctx, QueueSe long beginTimestampFirst = System.currentTimeMillis(); AddressableMessageQueue messageQueue = null; try { + if (messageList == null || messageList.isEmpty()) { + throw new ProxyException(ProxyExceptionCode.FORBIDDEN, "message list is empty"); + } Message message = messageList.get(0); String topic = message.getTopic(); if (isNeedCheckTopicMessageType(message)) { diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/processor/ProducerProcessorTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/processor/ProducerProcessorTest.java index e6a90df36be..2b8c3ef9a55 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/processor/ProducerProcessorTest.java +++ b/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.FORBIDDEN, 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.FORBIDDEN, cause.getCode()); + assertEquals("message list is empty", cause.getMessage()); + } + @Test public void testSendMessage() throws Throwable { when(metadataService.getTopicMessageType(any(), eq(TOPIC))).thenReturn(TopicMessageType.NORMAL); From d21c3e28712c9532308a337b139ed2269d0800ec Mon Sep 17 00:00:00 2001 From: liuhy Date: Tue, 4 Aug 2026 03:48:30 -0700 Subject: [PATCH 2/2] fix(proxy): classify empty producer requests --- .../proxy/common/ProxyExceptionCode.java | 1 + .../grpc/v2/common/GrpcProxyException.java | 1 + .../proxy/processor/ProducerProcessor.java | 2 +- .../activity/AbstractRemotingActivity.java | 1 + .../v2/common/GrpcProxyExceptionTest.java | 36 +++++++++++++++++++ .../processor/ProducerProcessorTest.java | 4 +-- 6 files changed, 42 insertions(+), 3 deletions(-) create mode 100644 proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcProxyExceptionTest.java diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/common/ProxyExceptionCode.java b/proxy/src/main/java/org/apache/rocketmq/proxy/common/ProxyExceptionCode.java index 4f91388215c..765a1e7b0d8 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/common/ProxyExceptionCode.java +++ b/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, diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcProxyException.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcProxyException.java index 74e499b4d72..f5c6e61a773 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcProxyException.java +++ b/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); diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ProducerProcessor.java b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ProducerProcessor.java index 881f0b8593f..bae5c872802 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ProducerProcessor.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/ProducerProcessor.java @@ -73,7 +73,7 @@ public CompletableFuture> sendMessage(ProxyContext ctx, QueueSe AddressableMessageQueue messageQueue = null; try { if (messageList == null || messageList.isEmpty()) { - throw new ProxyException(ProxyExceptionCode.FORBIDDEN, "message list is empty"); + throw new ProxyException(ProxyExceptionCode.INVALID_REQUEST, "message list is empty"); } Message message = messageList.get(0); String topic = message.getTopic(); diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/AbstractRemotingActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/AbstractRemotingActivity.java index 2d09c394299..81c67ec455c 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/AbstractRemotingActivity.java +++ b/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 PROXY_EXCEPTION_RESPONSE_CODE_MAP = new HashMap() { { 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); diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcProxyExceptionTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcProxyExceptionTest.java new file mode 100644 index 00000000000..192d37e4dbc --- /dev/null +++ b/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()); + } +} diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/processor/ProducerProcessorTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/processor/ProducerProcessorTest.java index 2b8c3ef9a55..24ff9d2acde 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/processor/ProducerProcessorTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/processor/ProducerProcessorTest.java @@ -91,7 +91,7 @@ public void testSendMessageRejectsEmptyMessageList() { assertTrue(exception.getCause() instanceof ProxyException); ProxyException cause = (ProxyException) exception.getCause(); - assertEquals(ProxyExceptionCode.FORBIDDEN, cause.getCode()); + assertEquals(ProxyExceptionCode.INVALID_REQUEST, cause.getCode()); assertEquals("message list is empty", cause.getMessage()); } @@ -110,7 +110,7 @@ public void testSendMessageRejectsNullMessageList() { assertTrue(exception.getCause() instanceof ProxyException); ProxyException cause = (ProxyException) exception.getCause(); - assertEquals(ProxyExceptionCode.FORBIDDEN, cause.getCode()); + assertEquals(ProxyExceptionCode.INVALID_REQUEST, cause.getCode()); assertEquals("message list is empty", cause.getMessage()); }