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 538a37ee fix(cluster): harden live cluster configuration and discovery
(#994)
538a37ee is described below
commit 538a37ee5fb76f4b93bd28e1d2152f0becc5bff2
Author: aias00 <[email protected]>
AuthorDate: Wed Aug 5 02:40:57 2026 -0700
fix(cluster): harden live cluster configuration and discovery (#994)
---
.../studio/cluster/broker/ClusterService.java | 27 +++++++++++---
.../studio/cluster/broker/ClusterServiceTest.java | 43 ++++++++++++++++++----
web/src/pages/cluster/index.tsx | 7 +++-
3 files changed, 63 insertions(+), 14 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterService.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterService.java
index d50c8499..e57b5cf8 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterService.java
@@ -52,7 +52,7 @@ public class ClusterService {
discovered.forEach(this::enrichWithLiveConfig);
return discovered;
}
- return clusterRepository.findAll();
+ return List.of();
}
public ClusterVO getCluster(String id) {
@@ -62,8 +62,7 @@ public class ClusterService {
enrichWithLiveConfig(live);
return live;
}
- return clusterRepository.findById(id)
- .orElseThrow(() -> new BusinessException(404, "Cluster not
found: " + id));
+ throw new BusinessException(503, "Cluster details are unavailable: " +
id);
}
/**
@@ -92,8 +91,8 @@ public class ClusterService {
public ClusterVO updateClusterConfig(UpdateConfigDTO command) {
log.info("Updating cluster config for: {}", command.getId());
- ClusterVO cluster = clusterRepository.findById(command.getId())
- .orElseThrow(() -> new BusinessException(404, "Cluster not
found: " + command.getId()));
+ requireMatchingDefaultQueueNums(command);
+ ClusterVO cluster = resolveCluster(command.getId());
ClusterConfigVO config = copyConfig(cluster.getConfig());
@@ -138,6 +137,16 @@ public class ClusterService {
return cluster;
}
+ private ClusterVO resolveCluster(String clusterId) {
+ ClusterVO live = clusterProvider.refreshClusterDetail(clusterId);
+ if (live != null) {
+ enrichWithLiveConfig(live);
+ return live;
+ }
+ return clusterRepository.findById(clusterId)
+ .orElseThrow(() -> new BusinessException(404, "Cluster not
found: " + clusterId));
+ }
+
private ClusterConfigVO copyConfig(ClusterConfigVO config) {
if (config == null) {
return new ClusterConfigVO();
@@ -185,6 +194,14 @@ public class ClusterService {
return props;
}
+ private void requireMatchingDefaultQueueNums(UpdateConfigDTO command) {
+ if (command.getWriteQueueNums() != null && command.getReadQueueNums()
!= null
+ &&
!command.getWriteQueueNums().equals(command.getReadQueueNums())) {
+ throw new BusinessException(400,
+ "RocketMQ broker default queue count requires matching
writeQueueNums and readQueueNums");
+ }
+ }
+
private FlushDiskType parseFlushDiskType(String value) {
try {
return FlushDiskType.valueOf(value);
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterServiceTest.java
index a8bd097f..f980f8f0 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterServiceTest.java
@@ -112,19 +112,34 @@ class ClusterServiceTest {
.build();
secondCluster.setId("cluster-2");
-
when(clusterRepository.findAll()).thenReturn(Arrays.asList(sampleCluster,
secondCluster));
+
when(clusterProvider.discoverClusters()).thenReturn(Arrays.asList(sampleCluster,
secondCluster));
List<ClusterVO> result = clusterService.listClusters();
assertThat(result).hasSize(2);
assertThat(result.get(0).getName()).isEqualTo("test-cluster");
assertThat(result.get(1).getName()).isEqualTo("second-cluster");
- verify(clusterRepository).findAll();
+ verify(clusterRepository, never()).findAll();
+ }
+
+ @Test
+ void
updateClusterConfigShouldRejectDifferentDefaultReadAndWriteQueueNums() {
+ UpdateConfigDTO command = UpdateConfigDTO.builder()
+ .id("cluster-1")
+ .writeQueueNums(8)
+ .readQueueNums(16)
+ .build();
+
+ assertThatThrownBy(() -> clusterService.updateClusterConfig(command))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("requires matching writeQueueNums and
readQueueNums");
+
+ verifyNoInteractions(clusterRepository, clusterProvider);
}
@Test
void listClustersShouldReturnEmptyListWhenNoClusters() {
- when(clusterRepository.findAll()).thenReturn(Collections.emptyList());
+
when(clusterProvider.discoverClusters()).thenReturn(Collections.emptyList());
List<ClusterVO> result = clusterService.listClusters();
@@ -133,7 +148,7 @@ class ClusterServiceTest {
@Test
void getClusterShouldReturnClusterWhenFound() {
-
when(clusterRepository.findById("cluster-1")).thenReturn(Optional.of(sampleCluster));
+
when(clusterProvider.refreshClusterDetail("cluster-1")).thenReturn(sampleCluster);
ClusterVO result = clusterService.getCluster("cluster-1");
@@ -146,12 +161,12 @@ class ClusterServiceTest {
@Test
void getClusterShouldThrowWhenNotFound() {
-
when(clusterRepository.findById("nonexistent")).thenReturn(Optional.empty());
+
when(clusterProvider.refreshClusterDetail("nonexistent")).thenReturn(null);
assertThatThrownBy(() -> clusterService.getCluster("nonexistent"))
.isInstanceOf(BusinessException.class)
- .hasMessageContaining("Cluster not found: nonexistent")
- .satisfies(ex -> assertThat(((BusinessException)
ex).getCode()).isEqualTo(404));
+ .hasMessageContaining("Cluster details are unavailable:
nonexistent")
+ .satisfies(ex -> assertThat(((BusinessException)
ex).getCode()).isEqualTo(503));
}
@Test
@@ -500,6 +515,20 @@ class ClusterServiceTest {
.satisfies(ex -> assertThat(((BusinessException)
ex).getCode()).isEqualTo(404));
}
+ @Test
+ void updateClusterConfigShouldUseLiveClusterWhenItIsNotPersisted() {
+
when(clusterProvider.refreshClusterDetail("cluster-1")).thenReturn(sampleCluster);
+ UpdateConfigDTO command = UpdateConfigDTO.builder()
+ .id("cluster-1")
+ .maxMessageSize(8_388_608)
+ .build();
+
+ ClusterVO result = clusterService.updateClusterConfig(command);
+
+
assertThat(result.getConfig().getMaxMessageSize()).isEqualTo(8_388_608);
+ verify(clusterRepository).updateConfig("cluster-1",
result.getConfig());
+ }
+
private void assertUnsupportedOperation(ThrowableAssert.ThrowingCallable
callable, String message) {
assertThatThrownBy(callable)
.isInstanceOf(BusinessException.class)
diff --git a/web/src/pages/cluster/index.tsx b/web/src/pages/cluster/index.tsx
index 5d873148..031e5bff 100644
--- a/web/src/pages/cluster/index.tsx
+++ b/web/src/pages/cluster/index.tsx
@@ -570,6 +570,10 @@ const ClusterPage = () => {
<Form.Item label={t('cluster.readQueues')} name="readQueueNums">
<InputNumber min={1} max={256} style={{ width: '100%' }} />
</Form.Item>
+ <Text type="secondary" style={{ display: 'block', marginTop:
-16, marginBottom: 16 }}>
+ RocketMQ Broker uses one default Topic queue count; read and
write values must
+ match.
+ </Text>
<Form.Item label={t('cluster.brokerPermission')}
name="brokerPermission">
<InputNumber min={0} max={7} style={{ width: '100%' }} />
</Form.Item>
@@ -764,8 +768,7 @@ const ClusterPage = () => {
.flatMap((c) =>
c.proxies
.filter((p) => {
- const matchSearch =
- !proxySearchText || searchText(p.addr).includes(proxySearchText);
+ const matchSearch = !proxySearchText ||
searchText(p.addr).includes(proxySearchText);
return matchSearch;
})
.map((p) => ({
