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 8c4907c588a..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 @@ -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.INVALID_REQUEST, "message list is empty"); + } Message message = messageList.get(0); String topic = message.getTopic(); if (isNeedCheckTopicMessageType(message)) { 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 e6a90df36be..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 @@ -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);