From 5ce7e013dbebf31138d124c43795bf446afe0cc6 Mon Sep 17 00:00:00 2001 From: qianye Date: Thu, 13 Aug 2026 19:12:18 +0800 Subject: [PATCH 1/2] [ISSUE #10924] Fix gRPC FIFO batch send ordering --- .../rocketmq/proxy/config/ProxyConfig.java | 12 ++ .../grpc/v2/producer/SendMessageActivity.java | 149 +++++++++++++----- .../v2/producer/SendMessageActivityTest.java | 104 ++++++++++++ .../service/mqclient/MQClientAPIExtTest.java | 10 ++ 4 files changed, 233 insertions(+), 42 deletions(-) diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/config/ProxyConfig.java b/proxy/src/main/java/org/apache/rocketmq/proxy/config/ProxyConfig.java index a7896c11e07..20a66f7f4c5 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/config/ProxyConfig.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/config/ProxyConfig.java @@ -116,6 +116,10 @@ public class ProxyConfig implements ConfigFile { * max message body size, 0 or negative number means no limit for proxy */ private int maxMessageSize = 4 * 1024 * 1024; + /** + * max message count in one batch send request + */ + private int batchSendMaxMsgNum = 4096; /** * if true, proxy will check message body size and reject msg if it's body is empty */ @@ -595,6 +599,14 @@ public void setMaxMessageSize(int maxMessageSize) { this.maxMessageSize = maxMessageSize; } + public int getBatchSendMaxMsgNum() { + return batchSendMaxMsgNum; + } + + public void setBatchSendMaxMsgNum(int batchSendMaxMsgNum) { + this.batchSendMaxMsgNum = batchSendMaxMsgNum; + } + public int getMaxUserPropertySize() { return maxUserPropertySize; } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivity.java index c0138cae7fa..dba01d0b7e4 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivity.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivity.java @@ -34,6 +34,7 @@ import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Set; import java.util.concurrent.CompletableFuture; import org.apache.commons.lang3.StringUtils; @@ -41,6 +42,7 @@ import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.common.message.MessageAccessor; import org.apache.rocketmq.common.message.MessageConst; +import org.apache.rocketmq.common.message.MessageDecoder; import org.apache.rocketmq.common.sysflag.MessageSysFlag; import org.apache.rocketmq.proxy.common.ProxyContext; import org.apache.rocketmq.proxy.config.ConfigurationManager; @@ -75,13 +77,15 @@ public CompletableFuture sendMessage(ProxyContext ctx, Send apache.rocketmq.v2.Message message = messageList.get(0); Resource topic = message.getTopic(); validateTopic(topic); + List messages = buildMessage(ctx, messageList, topic); + validateBatchMessages(messageList, messages); future = this.messagingProcessor.sendMessage( ctx, new SendMessageQueueSelector(request), topic.getName(), buildSysFlag(message), - buildMessage(ctx, request.getMessagesList(), topic) + messages ).thenApply(result -> convertToSendMessageResponse(ctx, request, result)); } catch (Throwable t) { future.completeExceptionally(t); @@ -103,6 +107,54 @@ protected List buildMessage(ProxyContext context, List protoMessageList, + List messageList) { + if (protoMessageList.size() <= 1) { + return; + } + + ProxyConfig config = ConfigurationManager.getProxyConfig(); + if (protoMessageList.size() > config.getBatchSendMaxMsgNum()) { + throw new GrpcProxyException(Code.MESSAGE_CORRUPTED, + "batch message count cannot exceed the max " + config.getBatchSendMaxMsgNum()); + } + + apache.rocketmq.v2.Message firstMessage = protoMessageList.get(0); + MessageType messageType = firstMessage.getSystemProperties().getMessageType(); + if (!MessageType.NORMAL.equals(messageType) && !MessageType.FIFO.equals(messageType)) { + throw new GrpcProxyException(Code.MESSAGE_PROPERTY_CONFLICT_WITH_TYPE, + "batch send only supports normal or FIFO messages"); + } + + String messageGroup = firstMessage.getSystemProperties().getMessageGroup(); + if (MessageType.FIFO.equals(messageType) && StringUtils.isBlank(messageGroup)) { + throw new GrpcProxyException(Code.ILLEGAL_MESSAGE_GROUP, + "message group cannot be empty for FIFO batch messages"); + } + Encoding bodyEncoding = firstMessage.getSystemProperties().getBodyEncoding(); + for (apache.rocketmq.v2.Message message : protoMessageList) { + if (!messageType.equals(message.getSystemProperties().getMessageType())) { + throw new GrpcProxyException(Code.MESSAGE_PROPERTY_CONFLICT_WITH_TYPE, + "all messages in a batch must have the same message type"); + } + if (!bodyEncoding.equals(message.getSystemProperties().getBodyEncoding())) { + throw new GrpcProxyException(Code.MESSAGE_CORRUPTED, + "all messages in a batch must have the same body encoding"); + } + if (MessageType.FIFO.equals(messageType) + && !Objects.equals(messageGroup, message.getSystemProperties().getMessageGroup())) { + throw new GrpcProxyException(Code.MESSAGE_PROPERTY_CONFLICT_WITH_TYPE, + "all FIFO messages in a batch must have the same message group"); + } + } + + int maxMessageSize = config.getMaxMessageSize(); + if (maxMessageSize > 0 && MessageDecoder.encodeMessages(messageList).length > maxMessageSize) { + throw new GrpcProxyException(Code.MESSAGE_BODY_TOO_LARGE, + "batch message body cannot exceed the max " + maxMessageSize); + } + } + protected Message buildMessage(ProxyContext context, apache.rocketmq.v2.Message protoMessage, String producerGroup) { String topicName = protoMessage.getTopic().getName(); @@ -333,41 +385,21 @@ protected SendMessageResponse convertToSendMessageResponse(ProxyContext ctx, Sen SendMessageResponse.Builder builder = SendMessageResponse.newBuilder(); Set responseCodes = new HashSet<>(); - for (SendResult result : resultList) { - SendResultEntry resultEntry; - switch (result.getSendStatus()) { - case FLUSH_DISK_TIMEOUT: - resultEntry = SendResultEntry.newBuilder() - .setStatus(ResponseBuilder.getInstance().buildStatus(Code.MASTER_PERSISTENCE_TIMEOUT, "send message failed, sendStatus=" + result.getSendStatus())) - .build(); - break; - case FLUSH_SLAVE_TIMEOUT: - resultEntry = SendResultEntry.newBuilder() - .setStatus(ResponseBuilder.getInstance().buildStatus(Code.SLAVE_PERSISTENCE_TIMEOUT, "send message failed, sendStatus=" + result.getSendStatus())) - .build(); - break; - case SLAVE_NOT_AVAILABLE: - resultEntry = SendResultEntry.newBuilder() - .setStatus(ResponseBuilder.getInstance().buildStatus(Code.HA_NOT_AVAILABLE, "send message failed, sendStatus=" + result.getSendStatus())) - .build(); - break; - case SEND_OK: - resultEntry = SendResultEntry.newBuilder() - .setStatus(ResponseBuilder.getInstance().buildStatus(Code.OK, Code.OK.name())) - .setOffset(result.getQueueOffset()) - .setMessageId(StringUtils.defaultString(result.getMsgId())) - .setTransactionId(StringUtils.defaultString(result.getTransactionId())) - .setRecallHandle(StringUtils.defaultString(result.getRecallHandle())) - .build(); - break; - default: - resultEntry = SendResultEntry.newBuilder() - .setStatus(ResponseBuilder.getInstance().buildStatus(Code.INTERNAL_SERVER_ERROR, "send message failed, sendStatus=" + result.getSendStatus())) - .build(); - break; + if (request.getMessagesCount() > 1 && resultList.size() == 1) { + SendResult batchResult = resultList.get(0); + for (int i = 0; i < request.getMessagesCount(); i++) { + SendResultEntry resultEntry = convertToSendResultEntry(batchResult, + request.getMessages(i).getSystemProperties().getMessageId(), batchResult.getQueueOffset() + i); + builder.addEntries(resultEntry); + responseCodes.add(resultEntry.getStatus().getCode()); + } + } else { + for (SendResult result : resultList) { + SendResultEntry resultEntry = convertToSendResultEntry(result, + StringUtils.defaultString(result.getMsgId()), result.getQueueOffset()); + builder.addEntries(resultEntry); + responseCodes.add(resultEntry.getStatus().getCode()); } - builder.addEntries(resultEntry); - responseCodes.add(resultEntry.getStatus().getCode()); } if (responseCodes.size() > 1) { builder.setStatus(ResponseBuilder.getInstance().buildStatus(Code.MULTIPLE_RESULTS, Code.MULTIPLE_RESULTS.name())); @@ -380,6 +412,42 @@ protected SendMessageResponse convertToSendMessageResponse(ProxyContext ctx, Sen return builder.build(); } + protected SendResultEntry convertToSendResultEntry(SendResult result, String messageId, long queueOffset) { + SendResultEntry resultEntry; + switch (result.getSendStatus()) { + case FLUSH_DISK_TIMEOUT: + resultEntry = SendResultEntry.newBuilder() + .setStatus(ResponseBuilder.getInstance().buildStatus(Code.MASTER_PERSISTENCE_TIMEOUT, "send message failed, sendStatus=" + result.getSendStatus())) + .build(); + break; + case FLUSH_SLAVE_TIMEOUT: + resultEntry = SendResultEntry.newBuilder() + .setStatus(ResponseBuilder.getInstance().buildStatus(Code.SLAVE_PERSISTENCE_TIMEOUT, "send message failed, sendStatus=" + result.getSendStatus())) + .build(); + break; + case SLAVE_NOT_AVAILABLE: + resultEntry = SendResultEntry.newBuilder() + .setStatus(ResponseBuilder.getInstance().buildStatus(Code.HA_NOT_AVAILABLE, "send message failed, sendStatus=" + result.getSendStatus())) + .build(); + break; + case SEND_OK: + resultEntry = SendResultEntry.newBuilder() + .setStatus(ResponseBuilder.getInstance().buildStatus(Code.OK, Code.OK.name())) + .setOffset(queueOffset) + .setMessageId(StringUtils.defaultString(messageId)) + .setTransactionId(StringUtils.defaultString(result.getTransactionId())) + .setRecallHandle(StringUtils.defaultString(result.getRecallHandle())) + .build(); + break; + default: + resultEntry = SendResultEntry.newBuilder() + .setStatus(ResponseBuilder.getInstance().buildStatus(Code.INTERNAL_SERVER_ERROR, "send message failed, sendStatus=" + result.getSendStatus())) + .build(); + break; + } + return resultEntry; + } + protected static class SendMessageQueueSelector implements QueueSelector { private final SendMessageRequest request; @@ -392,13 +460,10 @@ public SendMessageQueueSelector(SendMessageRequest request) { public AddressableMessageQueue select(ProxyContext ctx, MessageQueueView messageQueueView) { try { apache.rocketmq.v2.Message message = request.getMessages(0); - String shardingKey = null; - if (request.getMessagesCount() == 1) { - shardingKey = message.getSystemProperties().getMessageGroup(); - // lite topic - if (StringUtils.isBlank(shardingKey)) { - shardingKey = message.getSystemProperties().getLiteTopic(); - } + String shardingKey = message.getSystemProperties().getMessageGroup(); + // lite topic + if (StringUtils.isBlank(shardingKey)) { + shardingKey = message.getSystemProperties().getLiteTopic(); } AddressableMessageQueue targetMessageQueue; if (StringUtils.isNotEmpty(shardingKey)) { diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivityTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivityTest.java index f9761e299af..018a029f49f 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivityTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivityTest.java @@ -30,6 +30,7 @@ import com.google.protobuf.util.Timestamps; import java.time.Duration; import java.util.HashMap; +import java.util.List; import java.util.Map; import java.util.concurrent.CompletableFuture; import java.util.concurrent.ExecutionException; @@ -58,6 +59,7 @@ import org.assertj.core.util.Lists; import org.junit.Before; import org.junit.Test; +import org.mockito.ArgumentCaptor; import static org.apache.rocketmq.proxy.service.route.TopicRouteService.buildPenalizerByMQFaultStrategy; import static org.junit.Assert.assertEquals; @@ -68,6 +70,8 @@ import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; public class SendMessageActivityTest extends BaseActivityTest { @@ -122,6 +126,101 @@ public void sendMessage() throws Exception { assertEquals(msgId, response.getEntries(0).getMessageId()); } + @Test + @SuppressWarnings("unchecked") + public void testSendFifoBatchInOneInvocation() throws Exception { + String firstMessageId = MessageClientIDSetter.createUniqID(); + String secondMessageId = MessageClientIDSetter.createUniqID(); + String messageGroup = "group"; + SendResult sendResult = new SendResult(SendStatus.SEND_OK, null, null, null, 10); + when(this.messagingProcessor.sendMessage(any(), any(), anyString(), anyInt(), any())) + .thenReturn(CompletableFuture.completedFuture(Lists.newArrayList(sendResult))); + + SendMessageRequest request = SendMessageRequest.newBuilder() + .addMessages(createMessage(firstMessageId, MessageType.FIFO, messageGroup, 16)) + .addMessages(createMessage(secondMessageId, MessageType.FIFO, messageGroup, 16)) + .build(); + SendMessageResponse response = this.sendMessageActivity.sendMessage(createContext(), request).get(); + + ArgumentCaptor> messageListCaptor = + ArgumentCaptor.forClass(List.class); + verify(this.messagingProcessor, times(1)).sendMessage(any(), any(), anyString(), anyInt(), + messageListCaptor.capture()); + assertEquals(2, messageListCaptor.getValue().size()); + assertEquals(2, response.getEntriesCount()); + assertEquals(firstMessageId, response.getEntries(0).getMessageId()); + assertEquals(secondMessageId, response.getEntries(1).getMessageId()); + assertEquals(10, response.getEntries(0).getOffset()); + assertEquals(11, response.getEntries(1).getOffset()); + } + + @Test + public void testRejectFifoBatchWithDifferentMessageGroups() { + SendMessageRequest request = SendMessageRequest.newBuilder() + .addMessages(createMessage(MessageClientIDSetter.createUniqID(), MessageType.FIFO, "group-a", 16)) + .addMessages(createMessage(MessageClientIDSetter.createUniqID(), MessageType.FIFO, "group-b", 16)) + .build(); + + ExecutionException exception = assertThrows(ExecutionException.class, + () -> this.sendMessageActivity.sendMessage(createContext(), request).get()); + GrpcProxyException cause = (GrpcProxyException) exception.getCause(); + assertEquals(Code.MESSAGE_PROPERTY_CONFLICT_WITH_TYPE, cause.getCode()); + } + + @Test + public void testRejectBatchWhoseEncodedBodyExceedsLimit() { + int previousMaxMessageSize = ConfigurationManager.getProxyConfig().getMaxMessageSize(); + ConfigurationManager.getProxyConfig().setMaxMessageSize(80); + try { + List protoMessages = Lists.newArrayList( + createMessage(MessageClientIDSetter.createUniqID(), MessageType.NORMAL, "", 30), + createMessage(MessageClientIDSetter.createUniqID(), MessageType.NORMAL, "", 30)); + List messages = Lists.newArrayList( + new org.apache.rocketmq.common.message.Message(TOPIC, new byte[30]), + new org.apache.rocketmq.common.message.Message(TOPIC, new byte[30])); + + GrpcProxyException exception = assertThrows(GrpcProxyException.class, + () -> this.sendMessageActivity.validateBatchMessages(protoMessages, messages)); + assertEquals(Code.MESSAGE_BODY_TOO_LARGE, exception.getCode()); + } finally { + ConfigurationManager.getProxyConfig().setMaxMessageSize(previousMaxMessageSize); + } + } + + @Test + public void testRejectBatchWhoseMessageCountExceedsLimit() { + int previousMaxMessageCount = ConfigurationManager.getProxyConfig().getBatchSendMaxMsgNum(); + ConfigurationManager.getProxyConfig().setBatchSendMaxMsgNum(1); + try { + List protoMessages = Lists.newArrayList( + createMessage(MessageClientIDSetter.createUniqID(), MessageType.NORMAL, "", 1), + createMessage(MessageClientIDSetter.createUniqID(), MessageType.NORMAL, "", 1)); + List messages = Lists.newArrayList( + new org.apache.rocketmq.common.message.Message(TOPIC, new byte[1]), + new org.apache.rocketmq.common.message.Message(TOPIC, new byte[1])); + + GrpcProxyException exception = assertThrows(GrpcProxyException.class, + () -> this.sendMessageActivity.validateBatchMessages(protoMessages, messages)); + assertEquals(Code.MESSAGE_CORRUPTED, exception.getCode()); + } finally { + ConfigurationManager.getProxyConfig().setBatchSendMaxMsgNum(previousMaxMessageCount); + } + } + + private Message createMessage(String messageId, MessageType messageType, String messageGroup, int bodySize) { + return Message.newBuilder() + .setTopic(Resource.newBuilder().setName(TOPIC).build()) + .setSystemProperties(SystemProperties.newBuilder() + .setMessageId(messageId) + .setMessageType(messageType) + .setMessageGroup(messageGroup) + .setBornTimestamp(Timestamps.fromMillis(System.currentTimeMillis())) + .setBornHost(StringUtils.defaultString(NetworkUtil.getLocalAddress(), "127.0.0.1:1234")) + .build()) + .setBody(ByteString.copyFrom(new byte[bodySize])) + .build(); + } + @Test public void testConvertToSendMessageResponse() { { @@ -362,6 +461,11 @@ public void testSendOrderMessageQueueSelector() throws Exception { SendMessageActivity.SendMessageQueueSelector selector2 = new SendMessageActivity.SendMessageQueueSelector( SendMessageRequest.newBuilder() + .addMessages(Message.newBuilder() + .setSystemProperties(SystemProperties.newBuilder() + .setMessageGroup(String.valueOf(1)) + .build()) + .build()) .addMessages(Message.newBuilder() .setSystemProperties(SystemProperties.newBuilder() .setMessageGroup(String.valueOf(1)) diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/service/mqclient/MQClientAPIExtTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/service/mqclient/MQClientAPIExtTest.java index 1680c8732a6..2092c09049e 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/service/mqclient/MQClientAPIExtTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/service/mqclient/MQClientAPIExtTest.java @@ -54,6 +54,7 @@ import org.apache.rocketmq.remoting.RemotingClient; import org.apache.rocketmq.remoting.netty.NettyClientConfig; import org.apache.rocketmq.remoting.netty.ResponseFuture; +import org.apache.rocketmq.remoting.protocol.RequestCode; import org.apache.rocketmq.remoting.protocol.RemotingCommand; import org.apache.rocketmq.remoting.protocol.ResponseCode; import org.apache.rocketmq.remoting.protocol.body.GetLiteTopicInfoResponseBody; @@ -77,6 +78,7 @@ import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; +import org.mockito.ArgumentCaptor; import org.mockito.Mock; import org.mockito.Spy; import org.mockito.junit.MockitoJUnitRunner; @@ -91,6 +93,8 @@ import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; @RunWith(MockitoJUnitRunner.class) public class MQClientAPIExtTest { @@ -172,6 +176,12 @@ public void testSendMessageListAsync() throws Exception { assertNotNull(sendResult); assertEquals(sb.toString(), sendResult.getMsgId()); assertEquals(SendStatus.SEND_OK, sendResult.getSendStatus()); + + ArgumentCaptor requestCaptor = ArgumentCaptor.forClass(RemotingCommand.class); + verify(remotingClient, times(1)).invoke(anyString(), requestCaptor.capture(), anyLong()); + RemotingCommand request = requestCaptor.getValue(); + assertEquals(RequestCode.SEND_BATCH_MESSAGE, request.getCode()); + assertEquals(messageExtList.size(), MessageDecoder.decodeMessages(ByteBuffer.wrap(request.getBody())).size()); } @Test From e9633855b3d7490261434f0a930182662d826799 Mon Sep 17 00:00:00 2001 From: qianye Date: Thu, 13 Aug 2026 20:55:17 +0800 Subject: [PATCH 2/2] [ISSUE #10924] Reject compressed gRPC batch sends --- .../grpc/v2/producer/SendMessageActivity.java | 46 +++++++++++-------- .../v2/producer/SendMessageActivityTest.java | 31 +++++++++---- 2 files changed, 50 insertions(+), 27 deletions(-) diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivity.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivity.java index dba01d0b7e4..4f93a91507f 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivity.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivity.java @@ -39,6 +39,7 @@ import java.util.concurrent.CompletableFuture; import org.apache.commons.lang3.StringUtils; import org.apache.rocketmq.client.producer.SendResult; +import org.apache.rocketmq.common.attribute.TopicMessageType; import org.apache.rocketmq.common.message.Message; import org.apache.rocketmq.common.message.MessageAccessor; import org.apache.rocketmq.common.message.MessageConst; @@ -77,8 +78,9 @@ public CompletableFuture sendMessage(ProxyContext ctx, Send apache.rocketmq.v2.Message message = messageList.get(0); Resource topic = message.getTopic(); validateTopic(topic); + validateBatchEncoding(messageList); List messages = buildMessage(ctx, messageList, topic); - validateBatchMessages(messageList, messages); + validateBatchMessages(messages); future = this.messagingProcessor.sendMessage( ctx, @@ -107,42 +109,48 @@ protected List buildMessage(ProxyContext context, List protoMessageList, - List messageList) { - if (protoMessageList.size() <= 1) { + protected void validateBatchEncoding(List messageList) { + if (messageList.size() <= 1) { + return; + } + for (apache.rocketmq.v2.Message message : messageList) { + if (Encoding.GZIP.equals(message.getSystemProperties().getBodyEncoding())) { + throw new GrpcProxyException(Code.MESSAGE_CORRUPTED, + "batch send does not support compressed messages"); + } + } + } + + protected void validateBatchMessages(List messageList) { + if (messageList.size() <= 1) { return; } ProxyConfig config = ConfigurationManager.getProxyConfig(); - if (protoMessageList.size() > config.getBatchSendMaxMsgNum()) { + if (messageList.size() > config.getBatchSendMaxMsgNum()) { throw new GrpcProxyException(Code.MESSAGE_CORRUPTED, "batch message count cannot exceed the max " + config.getBatchSendMaxMsgNum()); } - apache.rocketmq.v2.Message firstMessage = protoMessageList.get(0); - MessageType messageType = firstMessage.getSystemProperties().getMessageType(); - if (!MessageType.NORMAL.equals(messageType) && !MessageType.FIFO.equals(messageType)) { + Message firstMessage = messageList.get(0); + TopicMessageType messageType = TopicMessageType.parseFromMessageProperty(firstMessage.getProperties()); + if (!TopicMessageType.NORMAL.equals(messageType) && !TopicMessageType.FIFO.equals(messageType)) { throw new GrpcProxyException(Code.MESSAGE_PROPERTY_CONFLICT_WITH_TYPE, "batch send only supports normal or FIFO messages"); } - String messageGroup = firstMessage.getSystemProperties().getMessageGroup(); - if (MessageType.FIFO.equals(messageType) && StringUtils.isBlank(messageGroup)) { + String messageGroup = firstMessage.getProperty(MessageConst.PROPERTY_SHARDING_KEY); + if (TopicMessageType.FIFO.equals(messageType) && StringUtils.isBlank(messageGroup)) { throw new GrpcProxyException(Code.ILLEGAL_MESSAGE_GROUP, "message group cannot be empty for FIFO batch messages"); } - Encoding bodyEncoding = firstMessage.getSystemProperties().getBodyEncoding(); - for (apache.rocketmq.v2.Message message : protoMessageList) { - if (!messageType.equals(message.getSystemProperties().getMessageType())) { + for (Message message : messageList) { + if (!messageType.equals(TopicMessageType.parseFromMessageProperty(message.getProperties()))) { throw new GrpcProxyException(Code.MESSAGE_PROPERTY_CONFLICT_WITH_TYPE, "all messages in a batch must have the same message type"); } - if (!bodyEncoding.equals(message.getSystemProperties().getBodyEncoding())) { - throw new GrpcProxyException(Code.MESSAGE_CORRUPTED, - "all messages in a batch must have the same body encoding"); - } - if (MessageType.FIFO.equals(messageType) - && !Objects.equals(messageGroup, message.getSystemProperties().getMessageGroup())) { + if (TopicMessageType.FIFO.equals(messageType) + && !Objects.equals(messageGroup, message.getProperty(MessageConst.PROPERTY_SHARDING_KEY))) { throw new GrpcProxyException(Code.MESSAGE_PROPERTY_CONFLICT_WITH_TYPE, "all FIFO messages in a batch must have the same message group"); } diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivityTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivityTest.java index 018a029f49f..fa1a0121a20 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivityTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/grpc/v2/producer/SendMessageActivityTest.java @@ -70,6 +70,7 @@ import static org.mockito.ArgumentMatchers.anyInt; import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -167,20 +168,37 @@ public void testRejectFifoBatchWithDifferentMessageGroups() { assertEquals(Code.MESSAGE_PROPERTY_CONFLICT_WITH_TYPE, cause.getCode()); } + @Test + public void testRejectCompressedBatch() { + Message compressedMessage = createMessage( + MessageClientIDSetter.createUniqID(), MessageType.NORMAL, "", 16); + compressedMessage = compressedMessage.toBuilder() + .setSystemProperties(compressedMessage.getSystemProperties().toBuilder() + .setBodyEncoding(Encoding.GZIP)) + .build(); + SendMessageRequest request = SendMessageRequest.newBuilder() + .addMessages(compressedMessage) + .addMessages(createMessage(MessageClientIDSetter.createUniqID(), MessageType.NORMAL, "", 16)) + .build(); + + ExecutionException exception = assertThrows(ExecutionException.class, + () -> this.sendMessageActivity.sendMessage(createContext(), request).get()); + GrpcProxyException cause = (GrpcProxyException) exception.getCause(); + assertEquals(Code.MESSAGE_CORRUPTED, cause.getCode()); + verify(this.messagingProcessor, never()).sendMessage(any(), any(), anyString(), anyInt(), any()); + } + @Test public void testRejectBatchWhoseEncodedBodyExceedsLimit() { int previousMaxMessageSize = ConfigurationManager.getProxyConfig().getMaxMessageSize(); ConfigurationManager.getProxyConfig().setMaxMessageSize(80); try { - List protoMessages = Lists.newArrayList( - createMessage(MessageClientIDSetter.createUniqID(), MessageType.NORMAL, "", 30), - createMessage(MessageClientIDSetter.createUniqID(), MessageType.NORMAL, "", 30)); List messages = Lists.newArrayList( new org.apache.rocketmq.common.message.Message(TOPIC, new byte[30]), new org.apache.rocketmq.common.message.Message(TOPIC, new byte[30])); GrpcProxyException exception = assertThrows(GrpcProxyException.class, - () -> this.sendMessageActivity.validateBatchMessages(protoMessages, messages)); + () -> this.sendMessageActivity.validateBatchMessages(messages)); assertEquals(Code.MESSAGE_BODY_TOO_LARGE, exception.getCode()); } finally { ConfigurationManager.getProxyConfig().setMaxMessageSize(previousMaxMessageSize); @@ -192,15 +210,12 @@ public void testRejectBatchWhoseMessageCountExceedsLimit() { int previousMaxMessageCount = ConfigurationManager.getProxyConfig().getBatchSendMaxMsgNum(); ConfigurationManager.getProxyConfig().setBatchSendMaxMsgNum(1); try { - List protoMessages = Lists.newArrayList( - createMessage(MessageClientIDSetter.createUniqID(), MessageType.NORMAL, "", 1), - createMessage(MessageClientIDSetter.createUniqID(), MessageType.NORMAL, "", 1)); List messages = Lists.newArrayList( new org.apache.rocketmq.common.message.Message(TOPIC, new byte[1]), new org.apache.rocketmq.common.message.Message(TOPIC, new byte[1])); GrpcProxyException exception = assertThrows(GrpcProxyException.class, - () -> this.sendMessageActivity.validateBatchMessages(protoMessages, messages)); + () -> this.sendMessageActivity.validateBatchMessages(messages)); assertEquals(Code.MESSAGE_CORRUPTED, exception.getCode()); } finally { ConfigurationManager.getProxyConfig().setBatchSendMaxMsgNum(previousMaxMessageCount);