(rocketmq-dashboard) branch rocketmq-studio updated: fix(instance): bound group enrichment by a batch deadline instead of per-group timeouts (#4877)

[lizhimin]

This is an automated email from the ASF dual-hosted git repository.

lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git


The following commit(s) were added to refs/heads/rocketmq-studio by this push:
     new 9fd4a1704 fix(instance): bound group enrichment by a batch deadline 
instead of per-group timeouts (#4877)
9fd4a1704 is described below

commit 9fd4a170424cd0747fd33d1eddf6834a1192142d
Author: Apulupie <167634780+frun...@users.noreply.github.com>
AuthorDate: Sat Sep 26 18:10:48 2026 +0800

    fix(instance): bound group enrichment by a batch deadline instead of 
per-group timeouts (#4877)
    
    Each enrichLiveStats future got its own three-second budget, so a hanging
    broker multiplied the listing wait by the number of groups. Give the whole
    batch one absolute deadline and cancel the remaining work once it passes,
    matching the InstanceResourceCountRunner approach.
---
 .../provider/apache/RocketMQMetadataProvider.java  | 15 ++++++++--
 .../apache/RocketMQMetadataProviderTest.java       | 32 ++++++++++++++++++++++
 2 files changed, 44 insertions(+), 3 deletions(-)

diff --git 
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
 
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
index 7f43cf5a1..a516abc77 100644
--- 
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
+++ 
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
@@ -329,11 +329,20 @@ public class RocketMQMetadataProvider implements 
MetadataProvider {
         for (ConsumerGroupVO vo : groups) {
             futures.add(onlineEnrichmentExecutor.submit(() -> 
enrichGroupLiveStats(instanceId, vo)));
         }
-        for (int i = 0; i < groups.size(); i++) {
+        // One absolute deadline for the whole batch (the 
InstanceResourceCountRunner
+        // precedent): a hanging broker must not multiply the wait by the 
number of
+        // groups, so each future only gets the time the batch has left.
+        long startedNanos = System.nanoTime();
+        long timeoutNanos = 
TimeUnit.SECONDS.toNanos(ONLINE_ENRICHMENT_TIMEOUT_SECONDS);
+        for (Future<?> future : futures) {
+            long remainingNanos = timeoutNanos - (System.nanoTime() - 
startedNanos);
             try {
-                futures.get(i).get(ONLINE_ENRICHMENT_TIMEOUT_SECONDS, 
TimeUnit.SECONDS);
+                if (remainingNanos <= 0) {
+                    throw new TimeoutException("enrichment batch deadline 
reached");
+                }
+                future.get(remainingNanos, TimeUnit.NANOSECONDS);
             } catch (TimeoutException e) {
-                futures.get(i).cancel(true);
+                future.cancel(true);
             } catch (InterruptedException e) {
                 Thread.currentThread().interrupt();
                 return;
diff --git 
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
 
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
index 5fb78a98c..308d55127 100644
--- 
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
+++ 
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
@@ -739,6 +739,38 @@ class RocketMQMetadataProviderTest {
                 .containsExactlyInAnyOrder("client-a", "client-b");
     }
 
+    @Test
+    void listConsumerGroupsShouldBoundEnrichmentToABatchDeadlineTest() throws 
Exception {
+        java.util.List<RmqGroup> entities = new java.util.ArrayList<>();
+        for (int i = 0; i < 3; i++) {
+            RmqGroup entity = new RmqGroup();
+            entity.setName("cg-slow-" + i);
+            entity.setInstanceId("instance-a");
+            entities.add(entity);
+        }
+        when(groupMapper.selectList(any())).thenReturn(entities);
+
+        DefaultMQAdminExt admin = 
org.mockito.Mockito.mock(DefaultMQAdminExt.class);
+        
when(admin.examineConsumerConnectionInfo(anyString())).thenAnswer(invocation -> 
{
+            Thread.sleep(10_000);
+            return new 
org.apache.rocketmq.remoting.protocol.body.ConsumerConnection();
+        });
+        when(runtimeAdminClientResolver.execute(eq("instance-a"), any()))
+                .thenAnswer(invocation ->
+                        
invocation.<MqAdminExtFactory.AdminAction<Object>>getArgument(1).apply(admin));
+
+        RocketMQMetadataProvider provider = newLiveProvider(admin);
+
+        long startedNanos = System.nanoTime();
+        provider.listConsumerGroups("instance-a", null, null);
+        long elapsedMillis = (System.nanoTime() - startedNanos) / 1_000_000L;
+
+        // The enrichment budget is one deadline for the whole batch (the
+        // InstanceResourceCountRunner precedent), not a per-group timeout that
+        // multiplies by the number of groups when a broker hangs.
+        assertThat(elapsedMillis).isLessThan(6_000L);
+    }
+
     @Test
     void 
listConsumerGroupsShouldMarkOnlineInstancesUnknownWhenConnectionLookupFailsTest()
 throws Exception {
         RmqGroup entity = new RmqGroup();