[rocketmq] branch develop updated: [ISSUE #7153] Add switch for MIXED message type (#7154)

[lo...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

lollipop pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git


The following commit(s) were added to refs/heads/develop by this push:
     new 8741ff8c9b [ISSUE #7153] Add switch for MIXED message type (#7154)
8741ff8c9b is described below

commit 8741ff8c9b3bdbfc97976285affa7ea35c81243c
Author: ShuangxiDing <di...@gmail.com>
AuthorDate: Thu Aug 10 17:41:15 2023 +0800

    [ISSUE #7153] Add switch for MIXED message type (#7154)
    
    Add a switch for MIXED message type when creating a Topic in the Broker.
---
 .../apache/rocketmq/broker/processor/AdminBrokerProcessor.java |  8 ++++++++
 .../src/main/java/org/apache/rocketmq/common/BrokerConfig.java | 10 ++++++++++
 2 files changed, 18 insertions(+)

diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
index a6ce03dc29..bbddcec2d7 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/AdminBrokerProcessor.java
@@ -59,6 +59,7 @@ import org.apache.rocketmq.common.TopicConfig;
 import org.apache.rocketmq.common.UnlockCallback;
 import org.apache.rocketmq.common.UtilAll;
 import org.apache.rocketmq.common.attribute.AttributeParser;
+import org.apache.rocketmq.common.attribute.TopicMessageType;
 import org.apache.rocketmq.common.constant.ConsumeInitMode;
 import org.apache.rocketmq.common.constant.FIleReadaheadMode;
 import org.apache.rocketmq.common.constant.LoggerName;
@@ -439,6 +440,13 @@ public class AdminBrokerProcessor implements NettyRequestProcessor {
         String attributesModification = requestHeader.getAttributes();
         topicConfig.setAttributes(AttributeParser.parseToMap(attributesModification));
 
+        if (topicConfig.getTopicMessageType() == TopicMessageType.MIXED
+            && !brokerController.getBrokerConfig().isEnableMixedMessageType()) {
+            response.setCode(ResponseCode.SYSTEM_ERROR);
+            response.setRemark("MIXED message type is not supported.");
+            return response;
+        }
+
         try {
             this.brokerController.getTopicConfigManager().updateTopicConfig(topicConfig);
             if (brokerController.getBrokerConfig().isEnableSingleTopicRegister()) {
diff --git a/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java b/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
index a815636b18..99a5db5ad9 100644
--- a/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
+++ b/common/src/main/java/org/apache/rocketmq/common/BrokerConfig.java
@@ -393,6 +393,8 @@ public class BrokerConfig extends BrokerIdentity {
      */
     private boolean enableSingleTopicRegister = false;
 
+    private boolean enableMixedMessageType = false;
+
     public long getMaxPopPollingSize() {
         return maxPopPollingSize;
     }
@@ -1712,4 +1714,12 @@ public class BrokerConfig extends BrokerIdentity {
     public void setEnableSingleTopicRegister(boolean enableSingleTopicRegister) {
         this.enableSingleTopicRegister = enableSingleTopicRegister;
     }
+
+    public boolean isEnableMixedMessageType() {
+        return enableMixedMessageType;
+    }
+
+    public void setEnableMixedMessageType(boolean enableMixedMessageType) {
+        this.enableMixedMessageType = enableMixedMessageType;
+    }
 }