Fix intermittent rocketmq-5-grpc-scenario test failure

simpleConsumeAsync() was fire-and-forget: it kicked off a
CompletableFuture chain (receiveAsync -> ackAsync) but returned
immediately without waiting for completion. The test thread would
exit and the HTTP response would be sent before consumer segments
were created, causing the mock collector to sometimes have fewer
than the expected 4 segments.

Changed from whenCompleteAsync (fire-and-forget) to thenComposeAsync
with .join(), so the method blocks until all async ack callbacks
have completed before returning.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: wu-sheng

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

@@ -203,26 +203,23 @@ public void simpleConsumeAsync(String topic,
                 maxMessageNum,
                 invisibleDuration
             );
-            future0.whenCompleteAsync((messages, throwable) -> {
-                if (null != throwable) {
-                    log.error("Failed to receive message from remote", throwable);
-                    return;
-                }
+            future0.thenComposeAsync(messages -> {
                 log.info("Received {} message(s)", messages.size());
                 final Map<MessageView, CompletableFuture<Void>> map =
                     messages.stream().collect(Collectors.toMap(message -> message, consumer::ackAsync));
-                for (Map.Entry<MessageView, CompletableFuture<Void>> entry : map.entrySet()) {
+                List<CompletableFuture<Void>> ackFutures = map.entrySet().stream().map(entry -> {
                     final MessageId messageId = entry.getKey().getMessageId();
                     final CompletableFuture<Void> future = entry.getValue();
-                    future.whenCompleteAsync((v, t) -> {
+                    return future.whenCompleteAsync((v, t) -> {
                         if (null != t) {
                             log.error("Message is failed to be acknowledged, messageId={}", messageId, t);
                             return;
                         }
                         log.info("Message is acknowledged successfully, messageId={}", messageId);
                     }, ackCallbackExecutor);
-                }
-            }, receiveCallbackExecutor);
+                }).collect(Collectors.toList());
+                return CompletableFuture.allOf(ackFutures.toArray(new CompletableFuture[0]));
+            }, receiveCallbackExecutor).join();
         } catch (Exception e) {
             log.error("consumer start error", e);
         }