From 77b5037bb8e6406a8a73bd7c1e379b660739b502 Mon Sep 17 00:00:00 2001 From: liuhy Date: Sun, 2 Aug 2026 23:46:53 -0700 Subject: [PATCH 1/4] [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 c70db9291a461cb726f9087e806199cd1fe5e646 Mon Sep 17 00:00:00 2001 From: liuhy Date: Tue, 4 Aug 2026 03:48:30 -0700 Subject: [PATCH 2/4] 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()); } From 4799dbc6b6d89762d6d87bc6d682dae4a064923c Mon Sep 17 00:00:00 2001 From: liuhy Date: Tue, 4 Aug 2026 04:01:02 -0700 Subject: [PATCH 3/4] fix(proxy): clarify invalid batch ack handles --- .../message/ClusterMessageService.java | 6 ++++ .../message/ClusterMessageServiceTest.java | 29 +++++++++++++++++++ 2 files changed, 35 insertions(+) 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..899d477dace 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 @@ -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 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(); + } } From d63c9faacaae91d82c475aadb15910a247669dd6 Mon Sep 17 00:00:00 2001 From: liuhy Date: Tue, 4 Aug 2026 03:44:26 -0700 Subject: [PATCH 4/4] fix(proxy): validate empty lock batch requests --- .../activity/ConsumerManagerActivity.java | 8 +- .../activity/ConsumerManagerActivityTest.java | 78 +++++++++++++++++++ 2 files changed, 84 insertions(+), 2 deletions(-) create mode 100644 proxy/src/test/java/org/apache/rocketmq/proxy/remoting/activity/ConsumerManagerActivityTest.java diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/ConsumerManagerActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/ConsumerManagerActivity.java index ce1f1b4a514..dcbb1b45b13 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/activity/ConsumerManagerActivity.java +++ b/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 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 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; } diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/remoting/activity/ConsumerManagerActivityTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/remoting/activity/ConsumerManagerActivityTest.java new file mode 100644 index 00000000000..521d228ba72 --- /dev/null +++ b/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()); + } +}