Fix rocketmq healthCheck re-entry when initialization exceeds curl timeout

Set consumerStarted flag at the start of the init block to prevent
concurrent healthCheck requests from re-entering. The initialization
(producer start + send + consumer start) can exceed curl's 3s timeout,
causing the health check to retry and re-enter the block repeatedly.

Author: wu-sheng

test/plugin/scenarios/rocketmq-5-grpc-scenario/src/main/java/test/apache/skywalking/apm/testcase/rocketmq/client/java/controller/CaseController.java

@@ -79,18 +79,26 @@ public String testcase() {
     @ResponseBody
     public String healthCheck() throws Exception {
         if (!consumerStarted) {
-            System.setProperty(MixAll.ROCKETMQ_HOME_ENV, this.getClass().getResource("/").getPath());
-            messageService.updateNormalTopic(NORMAL_TOPIC);
-            messageService.updateNormalTopic(ASYNC_PRODUCER_TOPIC);
-            messageService.updateNormalTopic(ASYNC_CONSUMER_TOPIC);
-            // Start push consumer early so it has time to complete rebalance
-            messageService.pushConsumes(
-                Collections.singletonList(NORMAL_TOPIC),
-                Collections.singletonList(TAG_NOMARL),
-                GROUP
-            );
-            final Producer producer = ProducerSingleton.getInstance(endpoints, NORMAL_TOPIC);
+            // Set flag early to prevent re-entry from concurrent healthCheck
+            // requests (each curl has a 3s timeout, and initialization may
+            // take longer than that).
             consumerStarted = true;
+            try {
+                System.setProperty(MixAll.ROCKETMQ_HOME_ENV, this.getClass().getResource("/").getPath());
+                messageService.updateNormalTopic(NORMAL_TOPIC);
+                messageService.updateNormalTopic(ASYNC_PRODUCER_TOPIC);
+                messageService.updateNormalTopic(ASYNC_CONSUMER_TOPIC);
+                // Start push consumer early so it has time to complete rebalance
+                messageService.pushConsumes(
+                    Collections.singletonList(NORMAL_TOPIC),
+                    Collections.singletonList(TAG_NOMARL),
+                    GROUP
+                );
+                final Producer producer = ProducerSingleton.getInstance(endpoints, NORMAL_TOPIC);
+            } catch (Exception e) {
+                consumerStarted = false;
+                throw e;
+            }
         }
 
         // Send probe messages until the consumer receives one, confirming

test/plugin/scenarios/rocketmq-scenario/src/main/java/test/apache/skywalking/apm/testcase/rocketmq/controller/CaseController.java

@@ -77,34 +77,42 @@ public String testcase() {
     @ResponseBody
     public String healthCheck() throws Exception {
         if (!consumerStarted) {
-            // Send a probe message to ensure the topic exists before consumer starts.
-            // Without this, the consumer's rebalance finds no queues and it would need
-            // another full rebalance cycle (~20s) after the topic is eventually created.
-            DefaultMQProducer probeProducer = new DefaultMQProducer("healthCheck_please_rename_unique_group_name");
-            probeProducer.setNamesrvAddr(namerServer);
-            probeProducer.start();
-            Message probeMsg = new Message("TopicTest", "probe".getBytes(StandardCharsets.UTF_8));
-            probeProducer.send(probeMsg);
-            probeProducer.shutdown();
-            System.out.printf("HealthCheck: Topic created via probe message.%n");
-
-            // Start consumer after topic exists. Use CONSUME_FROM_FIRST_OFFSET so
-            // the consumer picks up the probe message once rebalance completes.
-            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
-            consumer.setNamesrvAddr(namerServer);
-            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
-            consumer.subscribe("TopicTest", "*");
-            consumer.registerMessageListener(new MessageListenerConcurrently() {
-                @Override
-                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
-                    consumerReady = true;
-                    System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), new String(msgs.get(0).getBody(), StandardCharsets.UTF_8));
-                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
-                }
-            });
-            consumer.start();
+            // Set flag early to prevent re-entry from concurrent healthCheck
+            // requests (each curl has a 3s timeout, and initialization may
+            // take longer than that).
             consumerStarted = true;
-            System.out.printf("Consumer Started.%n");
+            try {
+                // Send a probe message to ensure the topic exists before consumer starts.
+                // Without this, the consumer's rebalance finds no queues and it would need
+                // another full rebalance cycle (~20s) after the topic is eventually created.
+                DefaultMQProducer probeProducer = new DefaultMQProducer("healthCheck_please_rename_unique_group_name");
+                probeProducer.setNamesrvAddr(namerServer);
+                probeProducer.start();
+                Message probeMsg = new Message("TopicTest", "probe".getBytes(StandardCharsets.UTF_8));
+                probeProducer.send(probeMsg);
+                probeProducer.shutdown();
+                System.out.printf("HealthCheck: Topic created via probe message.%n");
+
+                // Start consumer after topic exists. Use CONSUME_FROM_FIRST_OFFSET so
+                // the consumer picks up the probe message once rebalance completes.
+                DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name");
+                consumer.setNamesrvAddr(namerServer);
+                consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
+                consumer.subscribe("TopicTest", "*");
+                consumer.registerMessageListener(new MessageListenerConcurrently() {
+                    @Override
+                    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
+                        consumerReady = true;
+                        System.out.printf("%s Receive New Messages: %s %n", Thread.currentThread().getName(), new String(msgs.get(0).getBody(), StandardCharsets.UTF_8));
+                        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
+                    }
+                });
+                consumer.start();
+                System.out.printf("Consumer Started.%n");
+            } catch (Exception e) {
+                consumerStarted = false;
+                throw e;
+            }
         }
 
         // Wait until consumer has received the probe message, confirming