diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/common/ContextVariable.java b/proxy/src/main/java/org/apache/rocketmq/proxy/common/ContextVariable.java index fcc6bb02ff..f77ad376f4 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/common/ContextVariable.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/common/ContextVariable.java @@ -20,4 +20,5 @@ package org.apache.rocketmq.proxy.common; public class ContextVariable { public final static String REMOTE_ADDRESS = "remote-address"; public final static String LOCAL_ADDRESS = "local-address"; + public final static String MESSAGE_TYPE = "message-type"; } 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 2fbe49ef43..50a5ce1353 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 @@ -23,4 +23,5 @@ public enum ProxyExceptionCode { INVALID_RECEIPT_HANDLE, ILLEGAL_MESSAGE, INTERNAL_SERVER_ERROR, + TOPIC_MESSAGE_TYPE_NOT_MATCH, } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcConverter.java b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcConverter.java index 72db47eccf..04169146b0 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcConverter.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/grpc/v2/common/GrpcConverter.java @@ -35,6 +35,7 @@ import java.util.HashMap; import java.util.Map; import java.util.concurrent.TimeUnit; import org.apache.commons.lang3.StringUtils; +import org.apache.rocketmq.common.attribute.TopicMessageType; import org.apache.rocketmq.common.constant.LoggerName; import org.apache.rocketmq.common.consumer.ReceiptHandle; import org.apache.rocketmq.common.filter.ExpressionType; @@ -229,4 +230,18 @@ public class GrpcConverter { .setName(NamespaceUtil.withoutNamespace(resourceNameWithNamespace)) .build(); } + + public static TopicMessageType buildTopicMessageType(MessageType messageType) { + if (messageType.equals(MessageType.NORMAL)) { + return TopicMessageType.NORMAL; + } else if (messageType.equals(MessageType.FIFO)) { + return TopicMessageType.FIFO; + } else if (messageType.equals(MessageType.DELAY)) { + return TopicMessageType.DELAY; + } else if (messageType.equals(MessageType.TRANSACTION)) { + return TopicMessageType.TRANSACTION; + } else { + return TopicMessageType.UNSPECIFIED; + } + } } 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 998c0276fb..59bbb0ea0b 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 @@ -40,11 +40,15 @@ import java.util.Set; 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.MessageAccessor; import org.apache.rocketmq.common.message.MessageConst; import org.apache.rocketmq.common.message.MessageExt; import org.apache.rocketmq.common.sysflag.MessageSysFlag; +import org.apache.rocketmq.proxy.common.ContextVariable; import org.apache.rocketmq.proxy.common.ProxyContext; +import org.apache.rocketmq.proxy.common.ProxyException; +import org.apache.rocketmq.proxy.common.ProxyExceptionCode; import org.apache.rocketmq.proxy.grpc.v2.AbstractMessingActivity; import org.apache.rocketmq.proxy.grpc.v2.common.GrpcClientSettingsManager; import org.apache.rocketmq.proxy.grpc.v2.common.GrpcConverter; @@ -52,14 +56,18 @@ import org.apache.rocketmq.proxy.grpc.v2.common.GrpcProxyException; import org.apache.rocketmq.proxy.grpc.v2.common.ResponseBuilder; import org.apache.rocketmq.proxy.processor.MessagingProcessor; import org.apache.rocketmq.proxy.processor.QueueSelector; +import org.apache.rocketmq.proxy.processor.validator.DefaultTopicMessageTypeValidator; +import org.apache.rocketmq.proxy.processor.validator.TopicMessageTypeValidator; import org.apache.rocketmq.proxy.service.route.MessageQueueView; import org.apache.rocketmq.proxy.service.route.SelectableMessageQueue; public class SendMessageActivity extends AbstractMessingActivity { + private final TopicMessageTypeValidator validator; public SendMessageActivity(MessagingProcessor messagingProcessor, GrpcClientSettingsManager grpcClientSettingsManager) { super(messagingProcessor, grpcClientSettingsManager); + this.validator = new DefaultTopicMessageTypeValidator(); } public CompletableFuture sendMessage(Context ctx, SendMessageRequest request) { @@ -71,10 +79,13 @@ public class SendMessageActivity extends AbstractMessingActivity { throw new GrpcProxyException(Code.MESSAGE_CORRUPTED, "no message to send"); } + MessageType messageType = parseMessageType(request.getMessagesList()); + TopicMessageType topicMessageType = GrpcConverter.buildTopicMessageType(messageType); + List messageList = request.getMessagesList(); Resource topic = messageList.get(0).getTopic(); future = this.messagingProcessor.sendMessage( - context, + context.withVal(ContextVariable.MESSAGE_TYPE, topicMessageType.getValue()), new SendMessageQueueSelector(request), GrpcConverter.wrapResourceWithNamespace(topic), buildMessage(context, request.getMessagesList(), topic) @@ -241,6 +252,18 @@ public class SendMessageActivity extends AbstractMessingActivity { return builder.build(); } + protected MessageType parseMessageType(List messageList) { + MessageType messageType = messageList.get(0).getSystemProperties().getMessageType(); + for (Message message : messageList) { + MessageType messageType0 = message.getSystemProperties().getMessageType(); + if (messageType0 != messageType) { + throw new GrpcProxyException(Code.ILLEGAL_MESSAGE, "message type is not match"); + } + messageType = messageType0; + } + return messageType; + } + protected static class SendMessageQueueSelector implements QueueSelector { private final SendMessageRequest request; 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 a5baf4f9a5..b93187380a 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 @@ -23,6 +23,7 @@ import org.apache.commons.lang3.StringUtils; import org.apache.rocketmq.client.producer.SendResult; import org.apache.rocketmq.client.producer.SendStatus; import org.apache.rocketmq.common.MixAll; +import org.apache.rocketmq.common.attribute.TopicMessageType; import org.apache.rocketmq.common.consumer.ReceiptHandle; import org.apache.rocketmq.common.message.MessageAccessor; import org.apache.rocketmq.common.message.MessageConst; @@ -32,10 +33,14 @@ import org.apache.rocketmq.common.protocol.ResponseCode; import org.apache.rocketmq.common.protocol.header.ConsumerSendMsgBackRequestHeader; import org.apache.rocketmq.common.protocol.header.SendMessageRequestHeader; import org.apache.rocketmq.common.sysflag.MessageSysFlag; +import org.apache.rocketmq.proxy.common.ContextVariable; import org.apache.rocketmq.proxy.common.ProxyContext; import org.apache.rocketmq.proxy.common.ProxyException; import org.apache.rocketmq.proxy.common.ProxyExceptionCode; import org.apache.rocketmq.proxy.common.utils.FutureUtils; +import org.apache.rocketmq.proxy.config.ConfigurationManager; +import org.apache.rocketmq.proxy.processor.validator.DefaultTopicMessageTypeValidator; +import org.apache.rocketmq.proxy.processor.validator.TopicMessageTypeValidator; import org.apache.rocketmq.proxy.service.ServiceManager; import org.apache.rocketmq.proxy.service.route.SelectableMessageQueue; import org.apache.rocketmq.proxy.service.transaction.TransactionId; @@ -44,11 +49,13 @@ import org.apache.rocketmq.remoting.protocol.RemotingCommand; public class ProducerProcessor extends AbstractProcessor { private final ExecutorService executor; + private final TopicMessageTypeValidator topicMessageTypeValidator; public ProducerProcessor(MessagingProcessor messagingProcessor, ServiceManager serviceManager, ExecutorService executor) { super(messagingProcessor, serviceManager); this.executor = executor; + this.topicMessageTypeValidator = new DefaultTopicMessageTypeValidator(); } public CompletableFuture> sendMessage(ProxyContext ctx, QueueSelector queueSelector, @@ -56,6 +63,13 @@ public class ProducerProcessor extends AbstractProcessor { CompletableFuture> future = new CompletableFuture<>(); try { String topic = messageExtList.get(0).getTopic(); + if (ConfigurationManager.getProxyConfig().isEnableTopicMessageTypeCheck()) { + if (topicMessageTypeValidator != null) { + TopicMessageType topicMessageType = serviceManager.getMetadataService().getTopicMessageType(topic); + TopicMessageType messageType = TopicMessageType.valueOf(ctx.getVal(ContextVariable.MESSAGE_TYPE)); + topicMessageTypeValidator.validate(topicMessageType, messageType); + } + } SelectableMessageQueue messageQueue = queueSelector.select(ctx, this.serviceManager.getTopicRouteService().getCurrentMessageQueueView(topic)); if (messageQueue == null) { @@ -70,18 +84,18 @@ public class ProducerProcessor extends AbstractProcessor { messageExtList, requestHeader, timeoutMillis) - .thenApplyAsync(sendResultList -> { - for (SendResult sendResult : sendResultList) { - int tranType = MessageSysFlag.getTransactionValue(requestHeader.getSysFlag()); - if (SendStatus.SEND_OK.equals(sendResult.getSendStatus()) && - tranType == MessageSysFlag.TRANSACTION_PREPARED_TYPE && - StringUtils.isNotBlank(sendResult.getTransactionId())) { - TransactionId transactionId = TransactionId.genByBrokerTransactionId(messageQueue.getBrokerName(), sendResult); - sendResult.setTransactionId(transactionId.getProxyTransactionId()); + .thenApplyAsync(sendResultList -> { + for (SendResult sendResult : sendResultList) { + int tranType = MessageSysFlag.getTransactionValue(requestHeader.getSysFlag()); + if (SendStatus.SEND_OK.equals(sendResult.getSendStatus()) && + tranType == MessageSysFlag.TRANSACTION_PREPARED_TYPE && + StringUtils.isNotBlank(sendResult.getTransactionId())) { + TransactionId transactionId = TransactionId.genByBrokerTransactionId(messageQueue.getBrokerName(), sendResult); + sendResult.setTransactionId(transactionId.getProxyTransactionId()); + } } - } - return sendResultList; - }, this.executor); + return sendResultList; + }, this.executor); } catch (Throwable t) { future.completeExceptionally(t); } diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/processor/validator/DefaultTopicMessageTypeValidator.java b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/validator/DefaultTopicMessageTypeValidator.java new file mode 100644 index 0000000000..a0718c5ced --- /dev/null +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/validator/DefaultTopicMessageTypeValidator.java @@ -0,0 +1,31 @@ +/* + * 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.processor.validator; + +import org.apache.rocketmq.common.attribute.TopicMessageType; +import org.apache.rocketmq.proxy.common.ProxyException; +import org.apache.rocketmq.proxy.common.ProxyExceptionCode; + +public class DefaultTopicMessageTypeValidator implements TopicMessageTypeValidator { + + public void validate(TopicMessageType topicMessageType, TopicMessageType messageType) { + if (messageType.equals(TopicMessageType.UNSPECIFIED) || !messageType.equals(topicMessageType)) { + throw new ProxyException(ProxyExceptionCode.TOPIC_MESSAGE_TYPE_NOT_MATCH, messageType.name() + " " + topicMessageType.name()); + } + } +} diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/processor/validator/TopicMessageTypeValidator.java b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/validator/TopicMessageTypeValidator.java new file mode 100644 index 0000000000..43eae1e314 --- /dev/null +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/processor/validator/TopicMessageTypeValidator.java @@ -0,0 +1,29 @@ +/* + * 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.processor.validator; + +import org.apache.rocketmq.common.attribute.TopicMessageType; + +public interface TopicMessageTypeValidator { + /** + * Will throw {@link org.apache.rocketmq.proxy.common.ProxyException} if validate failed. + * @param topicMessageType Target topic + * @param messageType Message's type + */ + void validate(TopicMessageType topicMessageType, TopicMessageType messageType); +}