Fix rocketmq-5-grpc-scenario: avoid curl timeout by not joining consumer threads

The testcase() method was joining consumer threads that could block for
10-30 seconds, exceeding curl's 3-second timeout. This caused the main
HTTP entry span to never complete, so no producer segments were reported.

Fix: move PushConsumer startup to healthCheck() for early rebalance, and
start SimpleConsumer threads without joining so testcase() returns fast.
Background consumers complete within the 5-second post-curl sleep window.

Verified locally: both 5.1.1 and 5.1.4 pass with all expected segments.

Author: wu-sheng

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

@@ -53,30 +53,20 @@ public class CaseController {
     public String testcase() {
         try {
             messageService.sendNormalMessage(NORMAL_TOPIC, TAG_NOMARL, GROUP);
-            Thread t1 = new Thread(() -> messageService.pushConsumes(
-                Collections.singletonList(NORMAL_TOPIC),
-                Collections.singletonList(TAG_NOMARL),
-                GROUP
-            ));
-            t1.start();
-            t1.join();
 
             messageService.sendNormalMessageAsync(ASYNC_PRODUCER_TOPIC, TAG_ASYNC_PRODUCER, GROUP);
             messageService.sendNormalMessageAsync(ASYNC_PRODUCER_TOPIC, TAG_ASYNC_PRODUCER, GROUP);
-            Thread t2 = new Thread(() -> messageService.simpleConsumes(Collections.singletonList(ASYNC_PRODUCER_TOPIC),
-                                                           Collections.singletonList(TAG_ASYNC_PRODUCER), GROUP,
-                                                           10, 10
-            ));
-            t2.start();
-            t2.join();
+            new Thread(() -> messageService.simpleConsumes(
+                Collections.singletonList(ASYNC_PRODUCER_TOPIC),
+                Collections.singletonList(TAG_ASYNC_PRODUCER), GROUP,
+                10, 10
+            )).start();
 
             messageService.sendNormalMessage(ASYNC_CONSUMER_TOPIC, TAG_ASYNC_CONSUMER, GROUP);
             messageService.sendNormalMessage(ASYNC_CONSUMER_TOPIC, TAG_ASYNC_CONSUMER, GROUP);
-            Thread t3 = new Thread(() -> messageService.simpleConsumeAsync(ASYNC_CONSUMER_TOPIC, TAG_ASYNC_CONSUMER, GROUP, 10,
-                                                               10
-            ));
-            t3.start();
-            t3.join();
+            new Thread(() -> messageService.simpleConsumeAsync(
+                ASYNC_CONSUMER_TOPIC, TAG_ASYNC_CONSUMER, GROUP, 10, 10
+            )).start();
         } catch (Exception e) {
             log.error("testcase error", e);
         }
@@ -90,6 +80,12 @@ public String healthCheck() throws Exception {
         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);
         return SUCCESS;
     }