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 0b131e631 [ISSUE #2902][ISSUE #2903][ISSUE #2911][ISSUE #2912][ISSUE
#2913] fix(server): consolidate collection and registry robustness (#2907)
0b131e631 is described below
commit 0b131e631d7a440a11869a9f6a04543605caad33
Author: shown <[email protected]>
AuthorDate: Fri Sep 4 14:17:09 2026 +0800
[ISSUE #2902][ISSUE #2903][ISSUE #2911][ISSUE #2912][ISSUE #2913]
fix(server): consolidate collection and registry robustness (#2907)
* [ISSUE #2902] Run NameServer registry probes concurrently
Signed-off-by: yuluo-yx <[email protected]>
* [ISSUE #2903] Stabilize Apache provider broker fallback
* [ISSUE #2911] Fix Aliyun lag aggregation
* [ISSUE #2912] Align system group filtering
* [ISSUE #2913] Make metric snapshot batches atomic
---------
Signed-off-by: yuluo-yx <[email protected]>
---
.../studio/cluster/broker/RegistryProbeRunner.java | 4 +-
.../MybatisPlusMetricSnapshotRepository.java | 2 +
.../studio/common/util/SystemGroupFilter.java | 9 ++--
.../studio/provider/alibaba/AliyunConverters.java | 5 +-
.../provider/apache/RocketMQClusterProvider.java | 7 +--
.../cluster/broker/RegistryProbeRunnerTest.java | 44 ++++++++++++++++
.../MybatisPlusMetricSnapshotRepositoryTest.java | 10 ++++
.../studio/common/util/SystemGroupFilterTest.java | 12 +++--
.../provider/alibaba/AliyunConvertersLagTest.java | 60 ++++++++++++++++++++++
.../apache/RocketMQClusterProviderTest.java | 19 +++++++
10 files changed, 159 insertions(+), 13 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/RegistryProbeRunner.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/RegistryProbeRunner.java
index 16be981a6..22ea8bc58 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/RegistryProbeRunner.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/RegistryProbeRunner.java
@@ -74,8 +74,10 @@ class RegistryProbeRunner implements AutoCloseable {
* or failing entry contributes an empty result without affecting the
others.
*/
List<ClusterVO> probeAll(List<NameserverRegistryVO> entries, ProbeFunction
function) {
- return entries.stream()
+ List<CompletableFuture<List<ClusterVO>>> probes = entries.stream()
.map(entry -> probeOne(entry, function))
+ .toList();
+ return probes.stream()
.map(CompletableFuture::join)
.flatMap(List::stream)
.toList();
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/MybatisPlusMetricSnapshotRepository.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/MybatisPlusMetricSnapshotRepository.java
index 8cdd35480..2acc62368 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/MybatisPlusMetricSnapshotRepository.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/MybatisPlusMetricSnapshotRepository.java
@@ -24,6 +24,7 @@ import lombok.RequiredArgsConstructor;
import org.apache.rocketmq.studio.persistence.entity.RmqMetricSnapshot;
import org.apache.rocketmq.studio.persistence.mapper.RmqMetricSnapshotMapper;
import org.springframework.stereotype.Repository;
+import org.springframework.transaction.annotation.Transactional;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
@@ -42,6 +43,7 @@ public class MybatisPlusMetricSnapshotRepository implements
MetricSnapshotReposi
private final ObjectMapper objectMapper;
@Override
+ @Transactional
public void saveAll(List<MetricSample> samples) {
for (MetricSample sample : samples) {
mapper.insert(toEntity(sample));
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/common/util/SystemGroupFilter.java
b/server/src/main/java/org/apache/rocketmq/studio/common/util/SystemGroupFilter.java
index d352ec827..680deacb6 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/common/util/SystemGroupFilter.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/common/util/SystemGroupFilter.java
@@ -16,6 +16,8 @@
*/
package org.apache.rocketmq.studio.common.util;
+import org.apache.rocketmq.common.MixAll;
+
/**
* Shared utility for identifying RocketMQ system consumer groups.
*
@@ -39,15 +41,12 @@ public final class SystemGroupFilter {
if (group == null || group.isEmpty()) {
return true;
}
- return group.startsWith("CID_RMQ_SYS_")
+ return MixAll.isSysConsumerGroupPullMessage(group)
|| group.startsWith("CID_ONSAPI_")
|| group.startsWith("CID_SYS_")
|| group.startsWith("CID_HOUSEKEEPING")
|| group.startsWith("rmq_sys_")
|| group.startsWith("%RETRY%")
- || group.startsWith("%DLQ%")
- || group.startsWith("TOOLS_CONSUMER")
- || group.startsWith("FILTERSRV_CONSUMER")
- || group.startsWith("SELF_TEST_");
+ || group.startsWith("%DLQ%");
}
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConverters.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConverters.java
index 8fcfd06f2..fd52ed63b 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConverters.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConverters.java
@@ -192,7 +192,10 @@ final class AliyunConverters {
}
}
GetConsumerGroupLagResponseBody.TotalLag totalLag = data.getTotalLag();
- if (totalLag != null && totalLag.getReadyCount() != null) {
+ // The aggregate repeats the per-topic counts. Keep it only as a
fallback when
+ // the API does not expose the topic breakdown, otherwise callers that
sum rows
+ // report the same lag twice.
+ if (rows.isEmpty() && totalLag != null && totalLag.getReadyCount() !=
null) {
rows.add(QueueProgressVO.builder()
.broker("total")
.queueId(0)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProvider.java
index 8deca28eb..e6d12fbdf 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProvider.java
@@ -210,9 +210,10 @@ public class RocketMQClusterProvider implements
ClusterProvider {
// Use master address (brokerId = 0) preferentially
String masterAddr = brokerData.getBrokerAddrs().get(0L);
if (!StringUtils.hasText(masterAddr)) {
- masterAddr = brokerData.getBrokerAddrs().values().stream()
- .filter(StringUtils::hasText)
- .findFirst()
+ masterAddr = brokerData.getBrokerAddrs().entrySet().stream()
+ .filter(entry -> StringUtils.hasText(entry.getValue()))
+ .min(Map.Entry.comparingByKey())
+ .map(Map.Entry::getValue)
.orElse(null);
}
if (!StringUtils.hasText(masterAddr)) {
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/RegistryProbeRunnerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/RegistryProbeRunnerTest.java
new file mode 100644
index 000000000..92054cf6e
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/RegistryProbeRunnerTest.java
@@ -0,0 +1,44 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0.
+ */
+package org.apache.rocketmq.studio.cluster.broker;
+
+import org.apache.rocketmq.studio.cluster.nameserver.NameserverRegistryVO;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class RegistryProbeRunnerTest {
+
+ @Test
+ void startsIndependentRegistryProbesConcurrently() throws Exception {
+ CountDownLatch probesStarted = new CountDownLatch(2);
+ CountDownLatch releaseProbes = new CountDownLatch(1);
+ List<NameserverRegistryVO> entries = List.of(
+
NameserverRegistryVO.builder().name("registry-a").namesrvAddr("a:9876").build(),
+
NameserverRegistryVO.builder().name("registry-b").namesrvAddr("b:9876").build());
+
+ try (RegistryProbeRunner runner = new RegistryProbeRunner(2, 2,
2_000)) {
+ CompletableFuture<List<ClusterVO>> result =
CompletableFuture.supplyAsync(() ->
+ runner.probeAll(entries, entry -> {
+ probesStarted.countDown();
+ releaseProbes.await();
+ return
List.of(ClusterVO.builder().id(entry.getName()).build());
+ }));
+
+ assertThat(probesStarted.await(1, TimeUnit.SECONDS)).isTrue();
+ releaseProbes.countDown();
+ assertThat(result.get(1, TimeUnit.SECONDS))
+ .extracting(ClusterVO::getId)
+ .containsExactly("registry-a", "registry-b");
+ }
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/MybatisPlusMetricSnapshotRepositoryTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/MybatisPlusMetricSnapshotRepositoryTest.java
index 7d89fa996..76a8a5023 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/MybatisPlusMetricSnapshotRepositoryTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/MybatisPlusMetricSnapshotRepositoryTest.java
@@ -27,7 +27,9 @@ import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.transaction.annotation.Transactional;
+import java.lang.reflect.Method;
import java.time.Instant;
import java.util.List;
import java.util.Map;
@@ -43,6 +45,14 @@ class MybatisPlusMetricSnapshotRepositoryTest {
@Mock
private RmqMetricSnapshotMapper mapper;
+ @Test
+ void saveAllShouldUseOneTransactionForTheWholeSampleBatchTest() throws
Exception {
+ Method saveAll = MybatisPlusMetricSnapshotRepository.class
+ .getMethod("saveAll", List.class);
+
+ assertThat(saveAll.isAnnotationPresent(Transactional.class)).isTrue();
+ }
+
@Test
void nullClusterScopeShouldOnlyReadUnscopedSnapshotsTest() {
when(mapper.selectList(any(Wrapper.class))).thenReturn(List.of());
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/common/util/SystemGroupFilterTest.java
b/server/src/test/java/org/apache/rocketmq/studio/common/util/SystemGroupFilterTest.java
index 7068d46ed..3324949c5 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/common/util/SystemGroupFilterTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/common/util/SystemGroupFilterTest.java
@@ -33,10 +33,16 @@ class SystemGroupFilterTest {
assertThat(SystemGroupFilter.isSystem("rmq_sys_TRACE_DATA")).isTrue();
assertThat(SystemGroupFilter.isSystem("%RETRY%consumer-a")).isTrue();
assertThat(SystemGroupFilter.isSystem("%DLQ%consumer-a")).isTrue();
-
assertThat(SystemGroupFilter.isSystem("TOOLS_CONSUMER_monitor")).isTrue();
-
assertThat(SystemGroupFilter.isSystem("FILTERSRV_CONSUMER_filter")).isTrue();
- assertThat(SystemGroupFilter.isSystem("SELF_TEST_GROUP")).isTrue();
+ assertThat(SystemGroupFilter.isSystem("DEFAULT_CONSUMER")).isTrue();
+ assertThat(SystemGroupFilter.isSystem("TOOLS_CONSUMER")).isTrue();
+ assertThat(SystemGroupFilter.isSystem("SCHEDULE_CONSUMER")).isTrue();
+ assertThat(SystemGroupFilter.isSystem("FILTERSRV_CONSUMER")).isTrue();
+ assertThat(SystemGroupFilter.isSystem("__MONITOR_CONSUMER")).isTrue();
+ assertThat(SystemGroupFilter.isSystem("SELF_TEST_C_GROUP")).isTrue();
assertThat(SystemGroupFilter.isSystem("order-service-consumer")).isFalse();
+
assertThat(SystemGroupFilter.isSystem("TOOLS_CONSUMER_monitor")).isFalse();
+
assertThat(SystemGroupFilter.isSystem("FILTERSRV_CONSUMER_filter")).isFalse();
+ assertThat(SystemGroupFilter.isSystem("SELF_TEST_GROUP")).isFalse();
}
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConvertersLagTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConvertersLagTest.java
new file mode 100644
index 000000000..9fd14f418
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/alibaba/AliyunConvertersLagTest.java
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.rocketmq.studio.provider.alibaba;
+
+import com.aliyun.sdk.service.rocketmq20220801.models.DataTopicLagMapValue;
+import
com.aliyun.sdk.service.rocketmq20220801.models.GetConsumerGroupLagResponseBody;
+import org.apache.rocketmq.studio.instance.group.QueueProgressVO;
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class AliyunConvertersLagTest {
+
+ @Test
+ void topicBreakdownShouldNotBeCountedAgainAsAnAggregateRow() {
+ GetConsumerGroupLagResponseBody.Data data =
GetConsumerGroupLagResponseBody.Data.builder()
+ .topicLagMap(Map.of(
+ "orders",
DataTopicLagMapValue.builder().readyCount(40L).build(),
+ "payments",
DataTopicLagMapValue.builder().readyCount(60L).build()))
+
.totalLag(GetConsumerGroupLagResponseBody.TotalLag.builder().readyCount(100L).build())
+ .build();
+
+ List<QueueProgressVO> rows =
AliyunConverters.toQueueProgressRows(data);
+
+ assertThat(rows).extracting(QueueProgressVO::getTopic)
+ .containsExactlyInAnyOrder("orders", "payments");
+ assertThat(rows).noneMatch(row -> "total".equals(row.getBroker()));
+
assertThat(rows.stream().mapToLong(QueueProgressVO::getDiffTotal).sum()).isEqualTo(100L);
+ }
+
+ @Test
+ void aggregateShouldRemainAvailableWhenTopicBreakdownIsMissing() {
+ GetConsumerGroupLagResponseBody.Data data =
GetConsumerGroupLagResponseBody.Data.builder()
+
.totalLag(GetConsumerGroupLagResponseBody.TotalLag.builder().readyCount(100L).build())
+ .build();
+
+ assertThat(AliyunConverters.toQueueProgressRows(data)).singleElement()
+ .satisfies(row -> {
+ assertThat(row.getBroker()).isEqualTo("total");
+ assertThat(row.getDiffTotal()).isEqualTo(100L);
+ });
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProviderTest.java
index 89ea41124..bdfe2a6f2 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQClusterProviderTest.java
@@ -210,6 +210,25 @@ class RocketMQClusterProviderTest {
.isEqualTo("10.0.0.12:10911");
}
+ @Test
+ void discoverClustersShouldUseLowestBrokerIdWhenMasterIsMissing() throws
Exception {
+ DefaultMQAdminExt adminExt = mock(DefaultMQAdminExt.class);
+ RocketMQClusterProvider provider = newProvider(adminExt);
+ ClusterInfo info = clusterInfo();
+ HashMap<Long, String> addrs = new HashMap<>();
+ addrs.put(32L, "10.0.0.32:10911");
+ addrs.put(1L, "10.0.0.1:10911");
+ info.getBrokerAddrTable().put(
+ "broker-a", new BrokerData("DefaultCluster", "broker-a",
addrs));
+ when(adminExt.examineBrokerClusterInfo()).thenReturn(info);
+
+ ClusterVO cluster = provider.discoverClusters().get(0);
+
+ assertThat(cluster.getBrokers()).singleElement()
+ .extracting(broker -> broker.getAddr())
+ .isEqualTo("10.0.0.1:10911");
+ }
+
@Test
void discoverClustersShouldDiscoverProxiesViaHeartbeatSyncerTest() throws
Exception {
DefaultMQAdminExt adminExt = mock(DefaultMQAdminExt.class);
