Replace probe-based consumer readiness with 30s timer gate

The probe-based approach (waiting for consumer to receive a message)
works locally but fails in CI where the PushConsumer never receives
probe messages within the health check timeout. Switch to a simple
30-second timer gate after consumer startup, which reliably gives
enough time for rebalance to complete in both local and CI environments.

Author: wu-sheng

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

@@ -49,6 +49,7 @@ public class CaseController {
     private MessageService messageService;
 
     private volatile boolean consumerStarted = false;
+    private volatile long consumerStartTime = 0;
 
     @RequestMapping("/rocketmq-5-grpc-scenario")
     @ResponseBody
@@ -95,20 +96,17 @@ public String healthCheck() throws Exception {
                     GROUP
                 );
                 final Producer producer = ProducerSingleton.getInstance(endpoints, NORMAL_TOPIC);
+                consumerStartTime = System.currentTimeMillis();
             } catch (Exception e) {
                 consumerStarted = false;
                 throw e;
             }
         }
 
-        // Send probe messages until the consumer receives one, confirming
-        // that rebalance is complete and it can receive messages.
-        if (!MessageService.PUSH_CONSUMER_READY) {
-            try {
-                messageService.sendNormalMessage(NORMAL_TOPIC, TAG_NOMARL, GROUP);
-            } catch (Exception e) {
-                log.error("Probe message failed", e);
-            }
+        // PushConsumer needs time for initial rebalance with the broker.
+        // Return non-200 to force health check retries (3s each), giving
+        // the consumer enough time before the entry service is called.
+        if (System.currentTimeMillis() - consumerStartTime < 30000) {
             throw new RuntimeException("Consumer rebalance in progress");
         }
 

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

@@ -240,13 +240,10 @@ public void updateNormalTopic(String topic) {
         MQAdminStartup.main(subArgs);
     }
 
-    public static volatile boolean PUSH_CONSUMER_READY = false;
-
     public static class MyConsumer implements MessageListener {
 
         @Override
         public ConsumeResult consume(MessageView messageView) {
-            PUSH_CONSUMER_READY = true;
             log.info("Consume message successfully, messageId={},messageBody={}", messageView.getMessageId(),
                      messageView.getBody().toString()
             );

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

@@ -49,7 +49,7 @@ public class CaseController {
     private String namerServer;
 
     private volatile boolean consumerStarted = false;
-    private volatile boolean consumerReady = false;
+    private volatile long consumerStartTime = 0;
 
     @RequestMapping("/rocketmq-scenario")
     @ResponseBody
@@ -93,31 +93,31 @@ public String healthCheck() throws Exception {
                 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.
+                // Start consumer after topic exists so rebalance finds queues immediately.
                 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();
+                consumerStartTime = System.currentTimeMillis();
                 System.out.printf("Consumer Started.%n");
             } catch (Exception e) {
                 consumerStarted = false;
                 throw e;
             }
         }
 
-        // Wait until consumer has received the probe message, confirming
-        // that rebalance is complete and it can receive messages.
-        if (!consumerReady) {
+        // PushConsumer needs time for initial rebalance with the broker.
+        // Return non-200 to force health check retries (3s each), giving
+        // the consumer enough time before the entry service is called.
+        if (System.currentTimeMillis() - consumerStartTime < 30000) {
             throw new RuntimeException("Consumer rebalance in progress");
         }