[#1442] Support rocketmq message gray

pom.xml

@@ -54,6 +54,7 @@
     <module>spring-cloud-huawei-mesh</module>
     <module>spring-cloud-huawei-nacos</module>
     <module>spring-cloud-starter-huawei</module>
+    <module>spring-cloud-huawei-rocketmq</module>
   </modules>
 
   <licenses>

spring-cloud-huawei-bom/pom.xml

@@ -182,6 +182,11 @@
         <artifactId>spring-cloud-huawei-mesh</artifactId>
         <version>${project.version}</version>
       </dependency>
+      <dependency>
+        <groupId>com.huaweicloud</groupId>
+        <artifactId>spring-cloud-huawei-rocketmq</artifactId>
+        <version>${project.version}</version>
+      </dependency>
     </dependencies>
   </dependencyManagement>
 </project>

spring-cloud-huawei-dependencies/pom.xml

@@ -48,6 +48,8 @@
     <springdoc.version>2.7.0</springdoc.version>
     <nacos.client.version>2.3.0</nacos.client.version>
     <alibaba.context.support.version>1.0.11</alibaba.context.support.version>
+    <rocketmq.version>5.3.1</rocketmq.version>
+    <rocketmq.boot.version>2.3.3</rocketmq.boot.version>
   </properties>
 
   <dependencyManagement>
@@ -175,6 +177,21 @@
         <artifactId>spring-context-support</artifactId>
         <version>${alibaba.context.support.version}</version>
       </dependency>
+      <dependency>
+        <groupId>org.apache.rocketmq</groupId>
+        <artifactId>rocketmq-common</artifactId>
+        <version>${rocketmq.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>org.apache.rocketmq</groupId>
+        <artifactId>rocketmq-client</artifactId>
+        <version>${rocketmq.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>org.apache.rocketmq</groupId>
+        <artifactId>rocketmq-spring-boot</artifactId>
+        <version>${rocketmq.boot.version}</version>
+      </dependency>
     </dependencies>
   </dependencyManagement>
 </project>

spring-cloud-huawei-rocketmq/pom.xml

@@ -0,0 +1,59 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+  ~ Copyright (C) 2020-2024 Huawei Technologies Co., Ltd. All rights reserved.
+
+  ~ Licensed 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.
+  -->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>spring-cloud-huawei-parents</artifactId>
+        <groupId>com.huaweicloud</groupId>
+        <version>1.12.0-2024.0.x-SNAPSHOT</version>
+        <relativePath>../spring-cloud-huawei-parents</relativePath>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>spring-cloud-huawei-rocketmq</artifactId>
+    <name>Spring Cloud Huawei::Rocketmq</name>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.huaweicloud</groupId>
+            <artifactId>spring-cloud-huawei-common</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-web</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-common</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-client</artifactId>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.rocketmq</groupId>
+            <artifactId>rocketmq-spring-boot</artifactId>
+            <scope>provided</scope>
+        </dependency>
+    </dependencies>
+
+</project>

spring-cloud-huawei-rocketmq/src/main/java/com/huaweicloud/rocketmq/grayscale/ConditionalOnRocketMqMsgGrayEnabled.java

@@ -0,0 +1,31 @@
+/*
+
+ * Copyright (C) 2020-2025 Huawei Technologies Co., Ltd. All rights reserved.
+
+ * Licensed 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 com.huaweicloud.rocketmq.grayscale;
+
+import java.lang.annotation.ElementType;
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+
+@Retention(RetentionPolicy.RUNTIME)
+@Target({ElementType.TYPE, ElementType.METHOD})
+@ConditionalOnProperty(value = "rocketmq.gray.enabled", havingValue = "true")
+public @interface ConditionalOnRocketMqMsgGrayEnabled {
+}

spring-cloud-huawei-rocketmq/src/main/java/com/huaweicloud/rocketmq/grayscale/RocketMqMessageGrayConfiguration.java

@@ -0,0 +1,86 @@
+/*
+
+ * Copyright (C) 2020-2025 Huawei Technologies Co., Ltd. All rights reserved.
+
+ * Licensed 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 com.huaweicloud.rocketmq.grayscale;
+
+import org.springframework.boot.autoconfigure.AutoConfigureAfter;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.ApplicationContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.DependsOn;
+import org.springframework.core.env.Environment;
+
+import com.huaweicloud.rocketmq.grayscale.config.MessageGrayPropertiesManager;
+import com.huaweicloud.rocketmq.grayscale.servicemeta.ServiceMetaManager;
+import com.huaweicloud.rocketmq.grayscale.springboot.RocketMqMessageListenerAspect;
+import com.huaweicloud.rocketmq.grayscale.springboot.RocketMqListenerContainerAspect;
+import com.huaweicloud.rocketmq.grayscale.filter.WebMvcGrayHeaderFilter;
+import com.huaweicloud.rocketmq.grayscale.servicemeta.NacosServiceMeta;
+import com.huaweicloud.rocketmq.grayscale.servicemeta.ServicecombServiceMeta;
+import com.huaweicloud.rocketmq.grayscale.springboot.RocketMqSendMessageHookManager;
+
+@Configuration
+@ConditionalOnRocketMqMsgGrayEnabled
+@AutoConfigureAfter(MessageGrayPropertiesManager.class)
+public class RocketMqMessageGrayConfiguration {
+  @Bean
+  @ConfigurationProperties("spring.cloud.nacos.discovery")
+  public NacosServiceMeta serviceMetaInfo() {
+    return new NacosServiceMeta();
+  }
+
+  @Bean
+  @ConfigurationProperties("spring.cloud.servicecomb.instance")
+  public ServicecombServiceMeta servicecombServiceMeta() {
+    return new ServicecombServiceMeta();
+  }
+
+  @Bean
+  public MessageGrayPropertiesManager messageGrayPropertiesManager(Environment env) {
+    return new MessageGrayPropertiesManager(env);
+  }
+
+  @Bean
+  public ServiceMetaManager serviceMetaManager(NacosServiceMeta serviceMeta, ServicecombServiceMeta servicecombServiceMeta) {
+    return new ServiceMetaManager(serviceMeta, servicecombServiceMeta);
+  }
+
+  @Bean
+  @ConditionalOnClass(name = "org.apache.rocketmq.spring.support.DefaultRocketMQListenerContainer")
+  @DependsOn(value = {"messageGrayPropertiesManager", "serviceMetaManager"})
+  public RocketMqListenerContainerAspect rocketMQListenerContainerAspect() {
+    return new RocketMqListenerContainerAspect();
+  }
+
+  @Bean
+  @ConditionalOnClass(name = "org.apache.rocketmq.spring.core.RocketMQListener")
+  public RocketMqMessageListenerAspect rocketMqListenerAspect() {
+    return new RocketMqMessageListenerAspect();
+  }
+
+  @Bean
+  public RocketMqSendMessageHookManager rocketMqMessageHookManager(ApplicationContext applicationContext) {
+    return new RocketMqSendMessageHookManager(applicationContext);
+  }
+
+  @Bean
+  public WebMvcGrayHeaderFilter webMvcGrayHeaderFilter() {
+    return new WebMvcGrayHeaderFilter();
+  }
+}

spring-cloud-huawei-rocketmq/src/main/java/com/huaweicloud/rocketmq/grayscale/RocketMqMessageGrayUtils.java

@@ -0,0 +1,132 @@
+/*
+
+ * Copyright (C) 2020-2025 Huawei Technologies Co., Ltd. All rights reserved.
+
+ * Licensed 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 com.huaweicloud.rocketmq.grayscale;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.util.CollectionUtils;
+
+import com.huaweicloud.common.context.InvocationContext;
+import com.huaweicloud.common.context.InvocationContextHolder;
+import com.huaweicloud.rocketmq.grayscale.config.ConsumeModeEnum;
+import com.huaweicloud.rocketmq.grayscale.config.GrayscaleProperties;
+import com.huaweicloud.rocketmq.grayscale.config.RocketMqMessageGrayProperties;
+
+public class RocketMqMessageGrayUtils {
+  private static final Map<String, String> MICRO_SERVICE_PROPERTIES = new HashMap<>();
+
+  private static RocketMqMessageGrayProperties messageGrayProperties = new RocketMqMessageGrayProperties();
+
+  public static void setServiceMetaData(Map<String, String> serviceMeta) {
+    MICRO_SERVICE_PROPERTIES.putAll(serviceMeta);
+  }
+
+  public static void setMessageGrayProperties(RocketMqMessageGrayProperties grayProperties) {
+    messageGrayProperties = grayProperties;
+  }
+
+  public static String getGrayGroupTagsByServiceMeta() {
+    if (MICRO_SERVICE_PROPERTIES.isEmpty()) {
+      return "";
+    }
+    for (GrayscaleProperties item : messageGrayProperties.getGrayscale()) {
+      Map<String, String> ruleServiceMeta = item.getServiceMeta();
+      for (Map.Entry<String, String> entry : MICRO_SERVICE_PROPERTIES.entrySet()) {
+        if (ruleServiceMeta.containsKey(entry.getKey())
+            && StringUtils.equals(ruleServiceMeta.get(entry.getKey()), entry.getValue())) {
+          return item.getConsumerGroupTag();
+        }
+      }
+    }
+    return "";
+  }
+
+  public static RocketMqMessageGrayProperties getMessageGrayProperties() {
+    return messageGrayProperties;
+  }
+
+  public static long getAutoCheckDelayTime() {
+    return messageGrayProperties.getBase().getAutoCheckDelayTime();
+  }
+
+  public static ConsumeModeEnum getConsumeMode() {
+    return messageGrayProperties.getBase().getConsumeMode();
+  }
+
+  public static List<String> getExcludeGroupTags() {
+    return messageGrayProperties.getBase().getExcludeGroupTags();
+  }
+
+  public static Map<String, String> getGrayTagsByServiceMeta() {
+    return messageGrayProperties.getGrayTagsByServiceMeta(MICRO_SERVICE_PROPERTIES);
+  }
+
+  public static Map<String, String> getGrayTagsByGrayHeaders(Map<String, String> trafficGrayHeaders) {
+    return messageGrayProperties.getGrayTagsByGrayHeaders(trafficGrayHeaders);
+  }
+
+  public static Map<String, HashSet<String>> getAllTrafficTagMap() {
+    Map<String, HashSet<String>> trafficTags = new HashMap<>();
+    for (GrayscaleProperties grayscale : messageGrayProperties.getGrayscale()) {
+      for (String key : grayscale.getTrafficTag().keySet()) {
+        if (trafficTags.get(key) != null) {
+          trafficTags.get(key).add(grayscale.getTrafficTag().get(key));
+        } else {
+          HashSet<String> values = new HashSet<>();
+          values.add(grayscale.getTrafficTag().get(key));
+          trafficTags.put(key, values);
+        }
+      }
+    }
+    return trafficTags;
+  }
+
+  public static String getGrayConsumerGroup(String consumerGroup) {
+    String grayGroupTag = getGrayGroupTagsByServiceMeta();
+    if (StringUtils.isEmpty(grayGroupTag)) {
+      return consumerGroup;
+    }
+    String grayGroupSuffix = "_" + grayGroupTag;
+    if (consumerGroup.endsWith(grayGroupSuffix)) {
+      return consumerGroup;
+    }
+    return consumerGroup + grayGroupSuffix;
+  }
+
+  public static String buildCacheKey(String address, String topic, String consumerGroup) {
+    return address + "@" + topic + "@" + consumerGroup;
+  }
+
+  public static void setInvocationContext(Map<String, String> properties) {
+    Map<String, HashSet<String>> trafficTagMap = getAllTrafficTagMap();
+    if (CollectionUtils.isEmpty(properties) || CollectionUtils.isEmpty(trafficTagMap)) {
+      return;
+    }
+    InvocationContext invocationContext = InvocationContextHolder.getOrCreateInvocationContext();
+    for (Map.Entry<String, String> entry : properties.entrySet()) {
+      if (trafficTagMap.get(entry.getKey()) != null && entry.getValue() != null
+          && trafficTagMap.get(entry.getKey()).contains(entry.getValue())) {
+        invocationContext.putContext(entry.getKey(), entry.getValue());
+      }
+    }
+  }
+}