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/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/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/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); 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()); + } +} 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(); + } }