修复CI问题

sermant-io · Nov 7, 2024 · 5944b89 · 5944b89
1 parent 1148a90
commit 5944b89
Show file tree

Hide file tree

Showing 12 changed files with 87 additions and 50 deletions.
diff --git a/.github/actions/common/mq-grayscale/rocketmq/action.yml b/.github/actions/common/mq-grayscale/rocketmq/action.yml
@@ -56,4 +56,4 @@ runs:
     - name: entry
       uses: ./.github/actions/common/entry
       with:
-        log-dir: ./logs/rocketmq-grayscale/common
+        log-dir: ./logs/rocketmq-grayscale/common
diff --git a/.github/actions/common/plugin-change-check/action.yml b/.github/actions/common/plugin-change-check/action.yml
@@ -940,7 +940,7 @@ runs:
             ${{ steps.changed-common-action.outputs.changed }} == 'true' -o ${{ env.triggerPushEvent }} == 'true' ];then
          echo "enableXdsServicAction=true" >> $GITHUB_ENV
         fi
-        
+
         # ==========mq grayscale rocketmq is needed to test?==========
         if [ ${{ env.mqGrayscaleRocketMqChanged }} == 'true' -o \
             ${{ env.sermantAgentCoreChanged }} == 'true' -o \

diff --git a/.github/workflows/message_gray_integration_test.yml b/.github/workflows/message_gray_integration_test.yml
@@ -117,4 +117,4 @@ jobs:
       - name: common operations
         uses: ./.github/actions/common/mq-grayscale/rocketmq
       - name: message gray test for test-model=${{ matrix.test-model }} rocketMqClientVersion=${{ matrix.rocketMqClientVersion }}
-        uses: ./.github/actions/scenarios/mq-grayscale/rocketmq
+        uses: ./.github/actions/scenarios/mq-grayscale/rocketmq
diff --git a/...ant-integration-tests/mq-grayscale-rocketmq-test/grayscale-rocketmq-consumer-demo/pom.xml b/...ant-integration-tests/mq-grayscale-rocketmq-test/grayscale-rocketmq-consumer-demo/pom.xml
@@ -45,4 +45,5 @@
             </plugin>
         </plugins>
     </build>
+
 </project>
diff --git a/...src/main/java/io/sermant/demo/grayscale/rocketmq/consumer/RocketMqConsumerController.java b/...src/main/java/io/sermant/demo/grayscale/rocketmq/consumer/RocketMqConsumerController.java
@@ -16,22 +16,25 @@
 
 package io.sermant.demo.grayscale.rocketmq.consumer;
 
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-
 import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
 import org.apache.rocketmq.client.consumer.PullStatus;
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.client.impl.consumer.PullResultExt;
 import org.apache.rocketmq.common.message.MessageExt;
 import org.apache.rocketmq.common.message.MessageQueue;
+import org.apache.rocketmq.remoting.exception.RemotingException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RestController;
 
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
 /**
  * consumer controller
  *
@@ -52,33 +55,40 @@ public class RocketMqConsumerController {
     @Value("${rocketmq.topic}")
     private String topic;
 
-    @GetMapping("/getMessageCount")
-    public Map<String, Object> getMessageCount() {
-        return RocketMqMessageUtils.getMessageCount();
-    }
-
+    /**
+     * clear cache count info
+     *
+     * @return is success
+     */
     @GetMapping("/clearMessageCount")
     public String clearMessageCount() {
         RocketMqMessageUtils.clearMessageCount();
         return "success";
     }
 
-    @GetMapping("/consumePullMessage")
-    public Map<String, Object> consumePullMessage() {
+    /**
+     * pull message
+     *
+     * @return message count
+     */
+    @GetMapping("/getMessageCount")
+    public Map<String, Object> getMessageCount() {
         try {
             if (pullConsumer == null) {
                 pullConsumer = new DefaultMQPullConsumer("default");
                 pullConsumer.setNamesrvAddr(nameServer);
-                pullConsumer.setMaxReconsumeTimes(10000);
+                int maxReconsumeTimes = 10000;
+                pullConsumer.setMaxReconsumeTimes(maxReconsumeTimes);
                 pullConsumer.start();
             }
-            Set<MessageQueue> MessageQueues = pullConsumer.fetchSubscribeMessageQueues(topic);
-            if (MessageQueues.isEmpty()) {
+            Set<MessageQueue> messageQueues = pullConsumer.fetchSubscribeMessageQueues(topic);
+            if (messageQueues.isEmpty()) {
                 return new HashMap<>();
             }
-            for (MessageQueue mq : MessageQueues) {
+            int maxNums = 32;
+            for (MessageQueue mq : messageQueues) {
                 PullResultExt pullResult = (PullResultExt) pullConsumer.pullBlockIfNotFound(mq, null,
-                        getMessageQueueOffSet(mq), 32);
+                        getMessageQueueOffSet(mq), maxNums);
                 putMessageQueueOffSet(mq, pullResult);
                 if (PullStatus.FOUND == pullResult.getPullStatus()) {
                     List<MessageExt> messageExts = pullResult.getMsgFoundList();
@@ -88,7 +98,7 @@ public Map<String, Object> consumePullMessage() {
                 }
             }
             return RocketMqMessageUtils.getMessageCount();
-        } catch (Exception e) {
+        } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
             LOGGER.error("consumePullMessage error！", e);
             return new HashMap<>();
         }

diff --git a/...-demo/src/main/java/io/sermant/demo/grayscale/rocketmq/consumer/RocketMqMessageUtils.java b/...-demo/src/main/java/io/sermant/demo/grayscale/rocketmq/consumer/RocketMqMessageUtils.java
@@ -16,11 +16,11 @@
 
 package io.sermant.demo.grayscale.rocketmq.consumer;
 
+import org.apache.rocketmq.common.message.MessageExt;
+
 import java.util.HashMap;
 import java.util.Map;
 
-import org.apache.rocketmq.common.message.MessageExt;
-
 /**
  * message count utils
  *
@@ -34,6 +34,9 @@ public class RocketMqMessageUtils {
 
     private static int baseMessageCount = 0;
 
+    private RocketMqMessageUtils() {
+    }
+
     public static void convertMessageCount(MessageExt messageExt) {
         if (messageExt.getProperties() != null && GRAY.equals(messageExt.getProperties().get(
                 "x_lane_canary"))) {
@@ -43,12 +46,21 @@ public static void convertMessageCount(MessageExt messageExt) {
         baseMessageCount++;
     }
 
+    /**
+     * get base/gray message count
+     *
+     * @return message count
+     */
     public static Map<String, Object> getMessageCount() {
         Map<String, Object> countMap = new HashMap<>();
         countMap.put("baseMessageCount", baseMessageCount);
         countMap.put("grayMessageCount", grayMessageCount);
         return countMap;
     }
+
+    /**
+     * clear cache count info
+     */
     public static void clearMessageCount() {
         baseMessageCount = 0;
         grayMessageCount = 0;

diff --git a/...yscale-rocketmq-test/grayscale-rocketmq-consumer-demo/src/main/resources/application.yaml b/...yscale-rocketmq-test/grayscale-rocketmq-consumer-demo/src/main/resources/application.yaml
@@ -7,4 +7,4 @@ management:
   endpoints:
     web:
       exposure:
-        include: "*"
+        include: "*"
diff --git a/...t/src/test/java/io/sermant/demo/grayscale/rocketmq/integration/GrayscaleRocketmqTest.java b/...t/src/test/java/io/sermant/demo/grayscale/rocketmq/integration/GrayscaleRocketmqTest.java
@@ -133,11 +133,11 @@ public void testBaseBaseGrayExcGrayConsumeMessage() {
     }
 
     private String getBaseResult() {
-        return restTemplate.getForObject("http://127.0.0.1:9000/consumePullMessage", String.class);
+        return restTemplate.getForObject("http://127.0.0.1:9000/getMessageCount", String.class);
     }
 
     private String getGrayResult() {
-        return restTemplate.getForObject("http://127.0.0.1:9010/consumePullMessage", String.class);
+        return restTemplate.getForObject("http://127.0.0.1:9010/getMessageCount", String.class);
     }
 
     private void producerMessage() {
@@ -152,12 +152,12 @@ private void producerMessage() {
     private void triggerAndClearCacheCount(boolean isGrayInstanceInit) {
         if (isGrayInstanceInit) {
             // Trigger start gray consumer.
-            restTemplate.getForObject("http://127.0.0.1:9010/consumePullMessage", String.class);
+            restTemplate.getForObject("http://127.0.0.1:9010/getMessageCount", String.class);
             restTemplate.getForObject("http://127.0.0.1:9010/clearMessageCount", String.class);
         }
 
         // Trigger start base consumer.
-        restTemplate.getForObject("http://127.0.0.1:9000/consumePullMessage", String.class);
+        restTemplate.getForObject("http://127.0.0.1:9000/getMessageCount", String.class);
         restTemplate.getForObject("http://127.0.0.1:9000/clearMessageCount", String.class);
     }
 

diff --git a/...ant-integration-tests/mq-grayscale-rocketmq-test/grayscale-rocketmq-producer-demo/pom.xml b/...ant-integration-tests/mq-grayscale-rocketmq-test/grayscale-rocketmq-producer-demo/pom.xml
@@ -45,4 +45,5 @@
             </plugin>
         </plugins>
     </build>
+
 </project>
diff --git a/...-demo/src/main/java/io/sermant/demo/grayscale/rocketmq/producer/MqProducerController.java b/...-demo/src/main/java/io/sermant/demo/grayscale/rocketmq/producer/MqProducerController.java
@@ -16,15 +16,20 @@
 
 package io.sermant.demo.grayscale.rocketmq.producer;
 
+import org.apache.rocketmq.client.exception.MQBrokerException;
+import org.apache.rocketmq.client.exception.MQClientException;
 import org.apache.rocketmq.client.producer.DefaultMQProducer;
 import org.apache.rocketmq.common.message.Message;
+import org.apache.rocketmq.remoting.exception.RemotingException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.RestController;
 
+import java.nio.charset.StandardCharsets;
+
 /**
  * producer controller
  *
@@ -33,31 +38,38 @@
  */
 @RestController
 public class MqProducerController {
-  private static final Logger LOGGER = LoggerFactory.getLogger(MqProducerController.class);
+    private static final Logger LOGGER = LoggerFactory.getLogger(MqProducerController.class);
 
-  @Value("${rocketmq.address}")
-  private String mqAddress;
+    @Value("${rocketmq.address}")
+    private String mqAddress;
 
-  @Value("${rocketmq.topic}")
-  private String mqTopic;
+    @Value("${rocketmq.topic}")
+    private String mqTopic;
 
-  private DefaultMQProducer defaultMQProducer;
+    private DefaultMQProducer defaultMqProducer;
 
+  /**
+   * producer message
+   *
+   * @param message message
+   * @return is success
+   */
   @GetMapping("/producerMessage")
-  public String producerMessage(@RequestParam("message") String message) {
-    try {
-      if (defaultMQProducer == null) {
-        defaultMQProducer = new DefaultMQProducer("default");
-        defaultMQProducer.setNamesrvAddr(mqAddress);
-        defaultMQProducer.setSendMsgTimeout(60000);
-        defaultMQProducer.start();
-      }
-      Message mqMessage = new Message(mqTopic, message.getBytes());
-      defaultMQProducer.send(mqMessage);
-    } catch (Exception e) {
-      LOGGER.error("send message error, address={}, message={}", mqAddress, message, e);
-      return "error";
+    public String producerMessage(@RequestParam("message") String message) {
+        try {
+            if (defaultMqProducer == null) {
+                int sendMsgTimeout = 60000;
+                defaultMqProducer = new DefaultMQProducer("default");
+                defaultMqProducer.setNamesrvAddr(mqAddress);
+                defaultMqProducer.setSendMsgTimeout(sendMsgTimeout);
+                defaultMqProducer.start();
+            }
+            Message mqMessage = new Message(mqTopic, message.getBytes(StandardCharsets.UTF_8));
+            defaultMqProducer.send(mqMessage);
+        } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
+            LOGGER.error("send message error, address={}, message={}", mqAddress, message, e);
+            return "error";
+        }
+        return "success";
     }
-    return "success";
-  }
 }
diff --git a/...yscale-rocketmq-test/grayscale-rocketmq-producer-demo/src/main/resources/application.yaml b/...yscale-rocketmq-test/grayscale-rocketmq-producer-demo/src/main/resources/application.yaml
@@ -7,4 +7,4 @@ management:
   endpoints:
     web:
       exposure:
-        include: "*"
+        include: "*"
diff --git a/sermant-integration-tests/mq-grayscale-rocketmq-test/pom.xml b/sermant-integration-tests/mq-grayscale-rocketmq-test/pom.xml
@@ -57,4 +57,5 @@
         <module>grayscale-rocketmq-producer-demo</module>
         <module>grayscale-rocketmq-consumer-demo</module>
     </modules>
+
 </project>