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 1f374af9f feat(cluster): add broker configuration drift diagnostics
(#2746)
1f374af9f is described below
commit 1f374af9fc81a7ec758e8112b5dbfd8c2a987949
Author: coder999o <[email protected]>
AuthorDate: Mon Aug 31 20:32:29 2026 +0800
feat(cluster): add broker configuration drift diagnostics (#2746)
---
.../cluster/broker/BrokerConfigDiffService.java | 187 ++++++++++++++++
.../studio/cluster/broker/ClusterController.java | 9 +
.../studio/cluster/config/BrokerConfigDiffVO.java | 72 ++++++
.../broker/BrokerConfigDiffServiceTest.java | 243 +++++++++++++++++++++
.../cluster/broker/ClusterControllerTest.java | 61 ++++++
web/src/api/cluster.test.ts | 38 ++++
web/src/api/cluster.ts | 39 ++++
web/src/i18n/translations.ts | 29 +++
.../pages/cluster/__tests__/ClusterPage.test.tsx | 105 ++++++++-
web/src/pages/cluster/index.tsx | 206 ++++++++++++++++-
web/src/services/clusterService.ts | 54 +++++
11 files changed, 1040 insertions(+), 3 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/BrokerConfigDiffService.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/BrokerConfigDiffService.java
new file mode 100644
index 000000000..4096708af
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/BrokerConfigDiffService.java
@@ -0,0 +1,187 @@
+/*
+ * 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.cluster.broker;
+
+import org.apache.rocketmq.studio.cluster.config.BrokerConfigDiffVO;
+import org.apache.rocketmq.studio.cluster.config.ClusterConfigVO;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.provider.apache.RocketMQBrokerConfigService;
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Service;
+import org.springframework.util.StringUtils;
+
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+@Service
+@RequiredArgsConstructor
+public class BrokerConfigDiffService {
+
+ private static final List<ConfigField> COMPARED_FIELDS = List.of(
+ new ConfigField("flushDiskType", "flushDiskType"),
+ new ConfigField("autoCreateTopicEnable", "autoCreateTopicEnable"),
+ new ConfigField("autoCreateSubscriptionGroup",
"autoCreateSubscriptionGroup"),
+ new ConfigField("maxMessageSize", "maxMessageSize"),
+ new ConfigField("msgTraceTopicName", "msgTraceTopicName"),
+ new ConfigField("deleteWhen", "deleteWhen"),
+ new ConfigField("fileReservedTime", "fileReservedTime"),
+ new ConfigField("writeQueueNums", "defaultTopicQueueNums"),
+ new ConfigField("readQueueNums", "defaultTopicQueueNums"),
+ new ConfigField("brokerPermission", "brokerPermission"));
+
+ private final ClusterService clusterService;
+ private final RocketMQBrokerConfigService brokerConfigService;
+
+ public BrokerConfigDiffVO compare(String clusterId, String instanceId) {
+ String normalizedClusterId = requireClusterId(clusterId);
+ String normalizedInstanceId = normalizeInstanceId(instanceId);
+ ClusterVO cluster = normalizedInstanceId == null
+ ? clusterService.getCluster(normalizedClusterId)
+ : clusterService.getCluster(normalizedClusterId,
normalizedInstanceId);
+ List<BrokerTarget> brokers = collectBrokerTargets(cluster);
+ if (brokers.isEmpty()) {
+ throw new BusinessException(409, "Cluster has no broker endpoints:
" + normalizedClusterId);
+ }
+
+ Map<BrokerTarget, ClusterConfigVO> reachableConfigs = new
LinkedHashMap<>();
+ List<BrokerConfigDiffVO.BrokerStatusVO> statuses = new ArrayList<>();
+ for (BrokerTarget broker : brokers) {
+ try {
+ ClusterConfigVO config =
brokerConfigService.getBrokerConfig(broker.address(), normalizedInstanceId);
+ reachableConfigs.put(broker, config);
+ statuses.add(BrokerConfigDiffVO.BrokerStatusVO.builder()
+ .name(broker.name())
+ .address(broker.address())
+ .reachable(true)
+ .build());
+ } catch (BusinessException exception) {
+ statuses.add(BrokerConfigDiffVO.BrokerStatusVO.builder()
+ .name(broker.name())
+ .address(broker.address())
+ .reachable(false)
+ .build());
+ }
+ }
+
+ List<BrokerConfigDiffVO.ConfigDifferenceVO> differences =
findDifferences(reachableConfigs);
+ return BrokerConfigDiffVO.builder()
+ .cluster(normalizedClusterId)
+ .complete(reachableConfigs.size() == brokers.size())
+ .driftDetected(!differences.isEmpty())
+ .brokerCount(brokers.size())
+ .reachableBrokerCount(reachableConfigs.size())
+
.comparedFields(COMPARED_FIELDS.stream().map(ConfigField::field).toList())
+ .brokers(statuses)
+ .differences(differences)
+ .build();
+ }
+
+ private List<BrokerTarget> collectBrokerTargets(ClusterVO cluster) {
+ if (cluster.getBrokers() == null) {
+ return List.of();
+ }
+ Map<String, BrokerTarget> targets = new LinkedHashMap<>();
+ cluster.getBrokers().stream()
+ .filter(Objects::nonNull)
+ .filter(broker -> StringUtils.hasText(broker.getAddr()))
+ .forEach(broker -> {
+ String address = broker.getAddr().trim();
+ targets.putIfAbsent(address, new BrokerTarget(
+ StringUtils.hasText(broker.getName()) ?
broker.getName().trim() : address,
+ address));
+ });
+ return List.copyOf(targets.values());
+ }
+
+ private List<BrokerConfigDiffVO.ConfigDifferenceVO> findDifferences(
+ Map<BrokerTarget, ClusterConfigVO> configs) {
+ if (configs.size() < 2) {
+ return List.of();
+ }
+
+ List<BrokerConfigDiffVO.ConfigDifferenceVO> differences = new
ArrayList<>();
+ for (ConfigField field : COMPARED_FIELDS) {
+ List<BrokerConfigDiffVO.ConfigValueVO> values =
configs.entrySet().stream()
+ .map(entry -> configValue(entry.getKey(),
entry.getValue(), field))
+ .toList();
+ long distinctValues = values.stream()
+ .map(value -> value.isConfigured() ? value.getValue() :
null)
+ .distinct()
+ .count();
+ if (distinctValues > 1) {
+ differences.add(BrokerConfigDiffVO.ConfigDifferenceVO.builder()
+ .field(field.field())
+ .brokerProperty(field.brokerProperty())
+ .values(values)
+ .build());
+ }
+ }
+ return differences;
+ }
+
+ private BrokerConfigDiffVO.ConfigValueVO configValue(
+ BrokerTarget broker,
+ ClusterConfigVO config,
+ ConfigField field) {
+ String value = field.value(config);
+ return BrokerConfigDiffVO.ConfigValueVO.builder()
+ .brokerName(broker.name())
+ .address(broker.address())
+ .configured(value != null)
+ .value(value)
+ .build();
+ }
+
+ private String requireClusterId(String clusterId) {
+ if (!StringUtils.hasText(clusterId)) {
+ throw new BusinessException(400, "cluster is required");
+ }
+ return clusterId.trim();
+ }
+
+ private String normalizeInstanceId(String instanceId) {
+ return StringUtils.hasText(instanceId) ? instanceId.trim() : null;
+ }
+
+ private record BrokerTarget(String name, String address) {
+ }
+
+ private record ConfigField(String field, String brokerProperty) {
+ String value(ClusterConfigVO config) {
+ if (config == null) {
+ return null;
+ }
+ Object value = switch (field) {
+ case "flushDiskType" -> config.getFlushDiskType();
+ case "autoCreateTopicEnable" ->
config.isAutoCreateTopicEnable();
+ case "autoCreateSubscriptionGroup" ->
config.isAutoCreateSubscriptionGroup();
+ case "maxMessageSize" -> config.getMaxMessageSize();
+ case "msgTraceTopicName" -> config.getMsgTraceTopicName();
+ case "deleteWhen" -> config.getDeleteWhen();
+ case "fileReservedTime" -> config.getFileReservedTime();
+ case "writeQueueNums" -> config.getWriteQueueNums();
+ case "readQueueNums" -> config.getReadQueueNums();
+ case "brokerPermission" -> config.getBrokerPermission();
+ default -> null;
+ };
+ return value == null ? null : String.valueOf(value);
+ }
+ }
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterController.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterController.java
index 6797d779d..e53cfd7a4 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterController.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/broker/ClusterController.java
@@ -16,6 +16,7 @@
*/
package org.apache.rocketmq.studio.cluster.broker;
+import org.apache.rocketmq.studio.cluster.config.BrokerConfigDiffVO;
import org.apache.rocketmq.studio.cluster.config.ClusterConfigUpdateResultVO;
import org.apache.rocketmq.studio.cluster.config.ClusterConfigPreviewVO;
import org.apache.rocketmq.studio.cluster.config.UpdateConfigDTO;
@@ -42,6 +43,7 @@ public class ClusterController {
private final ClusterService clusterService;
private final ClusterConnectionService clusterConnectionService;
+ private final BrokerConfigDiffService brokerConfigDiffService;
@GetMapping
public Result<List<ClusterVO>> listClusters(@RequestParam(required =
false) String instanceId) {
@@ -76,6 +78,13 @@ public class ClusterController {
return Result.ok(clusterService.previewClusterConfig(command));
}
+ @GetMapping("/{id}/broker-config-diff")
+ public Result<BrokerConfigDiffVO> compareBrokerConfiguration(
+ @PathVariable String id,
+ @RequestParam(required = false) String instanceId) {
+ return Result.ok(brokerConfigDiffService.compare(id, instanceId));
+ }
+
@PostMapping("/{clusterId}/brokers/{name}/restart")
public Result<Map<String, Object>> restartBroker(@PathVariable String
clusterId,
@PathVariable String
name) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/config/BrokerConfigDiffVO.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/config/BrokerConfigDiffVO.java
new file mode 100644
index 000000000..d082e1dd4
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/config/BrokerConfigDiffVO.java
@@ -0,0 +1,72 @@
+/*
+ * 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.cluster.config;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.List;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class BrokerConfigDiffVO {
+
+ private String cluster;
+ private boolean complete;
+ private boolean driftDetected;
+ private int brokerCount;
+ private int reachableBrokerCount;
+ private List<String> comparedFields;
+ private List<BrokerStatusVO> brokers;
+ private List<ConfigDifferenceVO> differences;
+
+ @Data
+ @Builder
+ @NoArgsConstructor
+ @AllArgsConstructor
+ public static class BrokerStatusVO {
+ private String name;
+ private String address;
+ private boolean reachable;
+ private String message;
+ }
+
+ @Data
+ @Builder
+ @NoArgsConstructor
+ @AllArgsConstructor
+ public static class ConfigDifferenceVO {
+ private String field;
+ private String brokerProperty;
+ private List<ConfigValueVO> values;
+ }
+
+ @Data
+ @Builder
+ @NoArgsConstructor
+ @AllArgsConstructor
+ public static class ConfigValueVO {
+ private String brokerName;
+ private String address;
+ private boolean configured;
+ private String value;
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/BrokerConfigDiffServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/BrokerConfigDiffServiceTest.java
new file mode 100644
index 000000000..165f57e07
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/BrokerConfigDiffServiceTest.java
@@ -0,0 +1,243 @@
+/*
+ * 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.cluster.broker;
+
+import org.apache.rocketmq.studio.cluster.config.BrokerConfigDiffVO;
+import org.apache.rocketmq.studio.cluster.config.ClusterConfigVO;
+import org.apache.rocketmq.studio.common.domain.enums.FlushDiskType;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.provider.apache.RocketMQBrokerConfigService;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.Assertions.tuple;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+@ExtendWith(MockitoExtension.class)
+class BrokerConfigDiffServiceTest {
+
+ @Mock
+ private ClusterService clusterService;
+
+ @Mock
+ private RocketMQBrokerConfigService brokerConfigService;
+
+ private BrokerConfigDiffService service;
+
+ @BeforeEach
+ void setUp() {
+ service = new BrokerConfigDiffService(clusterService,
brokerConfigService);
+ }
+
+ @Test
+ void consistentBrokerConfigurationShouldBeReportedTest() {
+ when(clusterService.getCluster("cluster-a")).thenReturn(cluster(
+ broker("broker-a", "10.0.0.1:10911"),
+ broker("broker-b", "10.0.0.2:10911")));
+ when(brokerConfigService.getBrokerConfig("10.0.0.1:10911", null))
+ .thenReturn(config(FlushDiskType.ASYNC_FLUSH, true, 8, 6,
"04"));
+ when(brokerConfigService.getBrokerConfig("10.0.0.2:10911", null))
+ .thenReturn(config(FlushDiskType.ASYNC_FLUSH, true, 8, 6,
"04"));
+
+ BrokerConfigDiffVO result = service.compare(" cluster-a ", null);
+
+ assertThat(result.getCluster()).isEqualTo("cluster-a");
+ assertThat(result.isComplete()).isTrue();
+ assertThat(result.isDriftDetected()).isFalse();
+ assertThat(result.getBrokerCount()).isEqualTo(2);
+ assertThat(result.getReachableBrokerCount()).isEqualTo(2);
+ assertThat(result.getComparedFields())
+ .contains("flushDiskType", "autoCreateTopicEnable",
"writeQueueNums");
+ assertThat(result.getDifferences()).isEmpty();
+ assertThat(result.getBrokers())
+ .extracting(
+ BrokerConfigDiffVO.BrokerStatusVO::getName,
+ BrokerConfigDiffVO.BrokerStatusVO::getAddress,
+ BrokerConfigDiffVO.BrokerStatusVO::isReachable)
+ .containsExactly(
+ tuple("broker-a", "10.0.0.1:10911", true),
+ tuple("broker-b", "10.0.0.2:10911", true));
+ }
+
+ @Test
+ void changedBrokerValuesShouldBeExposedTest() {
+ when(clusterService.getCluster("cluster-a",
"instance-a")).thenReturn(cluster(
+ broker("broker-a", "10.0.0.1:10911"),
+ broker("broker-b", "10.0.0.2:10911")));
+ when(brokerConfigService.getBrokerConfig("10.0.0.1:10911",
"instance-a"))
+ .thenReturn(config(FlushDiskType.ASYNC_FLUSH, true, 8, 6,
"04"));
+ when(brokerConfigService.getBrokerConfig("10.0.0.2:10911",
"instance-a"))
+ .thenReturn(config(FlushDiskType.SYNC_FLUSH, false, 16, 4,
"06"));
+
+ BrokerConfigDiffVO result = service.compare(" cluster-a ", "
instance-a ");
+
+ assertThat(result.isComplete()).isTrue();
+ assertThat(result.isDriftDetected()).isTrue();
+ assertThat(result.getDifferences())
+ .extracting(BrokerConfigDiffVO.ConfigDifferenceVO::getField)
+ .contains(
+ "flushDiskType",
+ "autoCreateTopicEnable",
+ "writeQueueNums",
+ "readQueueNums",
+ "brokerPermission",
+ "deleteWhen");
+ BrokerConfigDiffVO.ConfigDifferenceVO queueNums =
result.getDifferences().stream()
+ .filter(difference ->
difference.getField().equals("writeQueueNums"))
+ .findFirst()
+ .orElseThrow();
+
assertThat(queueNums.getBrokerProperty()).isEqualTo("defaultTopicQueueNums");
+ assertThat(queueNums.getValues())
+ .extracting(
+ BrokerConfigDiffVO.ConfigValueVO::getBrokerName,
+ BrokerConfigDiffVO.ConfigValueVO::getAddress,
+ BrokerConfigDiffVO.ConfigValueVO::isConfigured,
+ BrokerConfigDiffVO.ConfigValueVO::getValue)
+ .containsExactly(
+ tuple("broker-a", "10.0.0.1:10911", true, "8"),
+ tuple("broker-b", "10.0.0.2:10911", true, "16"));
+ verify(clusterService).getCluster("cluster-a", "instance-a");
+ }
+
+ @Test
+ void partialResultsShouldBeKeptWhenBrokerConfigReadFailsTest() {
+ when(clusterService.getCluster("cluster-a")).thenReturn(cluster(
+ broker("broker-a", "10.0.0.1:10911"),
+ broker("broker-b", "10.0.0.2:10911")));
+ when(brokerConfigService.getBrokerConfig("10.0.0.1:10911", null))
+ .thenReturn(config(FlushDiskType.ASYNC_FLUSH, true, 8, 6,
"04"));
+ when(brokerConfigService.getBrokerConfig("10.0.0.2:10911", null))
+ .thenThrow(new BusinessException(502, "broker unavailable"));
+
+ BrokerConfigDiffVO result = service.compare("cluster-a", null);
+
+ assertThat(result.isComplete()).isFalse();
+ assertThat(result.isDriftDetected()).isFalse();
+ assertThat(result.getReachableBrokerCount()).isEqualTo(1);
+ assertThat(result.getBrokers())
+ .extracting(
+ BrokerConfigDiffVO.BrokerStatusVO::getAddress,
+ BrokerConfigDiffVO.BrokerStatusVO::isReachable,
+ BrokerConfigDiffVO.BrokerStatusVO::getMessage)
+ .containsExactly(
+ tuple("10.0.0.1:10911", true, null),
+ tuple("10.0.0.2:10911", false, null));
+ }
+
+ @Test
+ void brokerAddressesShouldBeDeduplicatedBeforeReadingConfigTest() {
+ when(clusterService.getCluster("cluster-a")).thenReturn(cluster(
+ broker("broker-a", "10.0.0.1:10911"),
+ broker("broker-a-duplicate", " 10.0.0.1:10911 "),
+ broker("broker-b", "10.0.0.2:10911")));
+ when(brokerConfigService.getBrokerConfig("10.0.0.1:10911", null))
+ .thenReturn(config(FlushDiskType.ASYNC_FLUSH, true, 8, 6,
"04"));
+ when(brokerConfigService.getBrokerConfig("10.0.0.2:10911", null))
+ .thenReturn(config(FlushDiskType.ASYNC_FLUSH, true, 8, 6,
"04"));
+
+ BrokerConfigDiffVO result = service.compare("cluster-a", null);
+
+ assertThat(result.getBrokerCount()).isEqualTo(2);
+ assertThat(result.getBrokers())
+ .extracting(BrokerConfigDiffVO.BrokerStatusVO::getAddress)
+ .containsExactly("10.0.0.1:10911", "10.0.0.2:10911");
+ }
+
+ @Test
+ void singleReachableBrokerShouldBeCompleteWithoutDriftTest() {
+ when(clusterService.getCluster("cluster-a")).thenReturn(cluster(
+ broker(null, "10.0.0.1:10911")));
+ when(brokerConfigService.getBrokerConfig("10.0.0.1:10911", null))
+ .thenReturn(config(FlushDiskType.ASYNC_FLUSH, true, 8, 6,
"04"));
+
+ BrokerConfigDiffVO result = service.compare("cluster-a", null);
+
+ assertThat(result.isComplete()).isTrue();
+ assertThat(result.isDriftDetected()).isFalse();
+ assertThat(result.getBrokerCount()).isEqualTo(1);
+ assertThat(result.getBrokers()).singleElement().satisfies(status -> {
+ assertThat(status.getName()).isEqualTo("10.0.0.1:10911");
+ assertThat(status.isReachable()).isTrue();
+ });
+ }
+
+ @Test
+ void clusterWithoutBrokerAddressesShouldBeRejectedTest() {
+ when(clusterService.getCluster("cluster-a")).thenReturn(cluster(
+ broker("broker-a", " "),
+ broker("broker-b", null)));
+
+ assertThatThrownBy(() -> service.compare("cluster-a", null))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Cluster has no broker endpoints: cluster-a")
+ .satisfies(exception -> assertThat(((BusinessException)
exception).getCode())
+ .isEqualTo(409));
+ }
+
+ @Test
+ void blankClusterIdShouldBeRejectedTest() {
+ assertThatThrownBy(() -> service.compare(" ", null))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("cluster is required")
+ .satisfies(exception -> assertThat(((BusinessException)
exception).getCode())
+ .isEqualTo(400));
+ }
+
+ private ClusterVO cluster(BrokerVO... brokers) {
+ ClusterVO cluster = ClusterVO.builder()
+ .name("cluster-a")
+ .brokers(List.of(brokers))
+ .build();
+ cluster.setId("cluster-a");
+ return cluster;
+ }
+
+ private BrokerVO broker(String name, String address) {
+ return BrokerVO.builder()
+ .name(name)
+ .addr(address)
+ .build();
+ }
+
+ private ClusterConfigVO config(
+ FlushDiskType flushDiskType,
+ boolean autoCreateTopic,
+ int queueNums,
+ int brokerPermission,
+ String deleteWhen) {
+ return ClusterConfigVO.builder()
+ .flushDiskType(flushDiskType)
+ .autoCreateTopicEnable(autoCreateTopic)
+ .autoCreateSubscriptionGroup(true)
+ .maxMessageSize(4194304)
+ .msgTraceTopicName("RMQ_SYS_TRACE_TOPIC")
+ .deleteWhen(deleteWhen)
+ .fileReservedTime(72)
+ .writeQueueNums(queueNums)
+ .readQueueNums(queueNums)
+ .brokerPermission(brokerPermission)
+ .build();
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterControllerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterControllerTest.java
index d4704080a..4765ea57a 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterControllerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/broker/ClusterControllerTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.rocketmq.studio.cluster.broker;
+import org.apache.rocketmq.studio.cluster.config.BrokerConfigDiffVO;
import org.apache.rocketmq.studio.cluster.config.ClusterConfigUpdateResultVO;
import org.apache.rocketmq.studio.cluster.config.ClusterConfigPreviewVO;
import org.apache.rocketmq.studio.cluster.config.ClusterConfigVO;
@@ -67,6 +68,9 @@ class ClusterControllerTest {
@MockBean
private ClusterConnectionService clusterConnectionService;
+ @MockBean
+ private BrokerConfigDiffService brokerConfigDiffService;
+
@Test
void listRegistryClustersShouldReturnDiscoveredClustersTest() throws
Exception {
when(clusterService.listRegistryClusters()).thenReturn(Collections.singletonList(
@@ -260,6 +264,63 @@ class ClusterControllerTest {
verify(clusterService).previewClusterConfig(any(UpdateConfigDTO.class));
}
+ @Test
+ void brokerConfigCompareShouldReturnDriftResultTest() throws Exception {
+ when(brokerConfigDiffService.compare("cluster-1",
"instance-1")).thenReturn(
+ BrokerConfigDiffVO.builder()
+ .cluster("cluster-1")
+ .complete(true)
+ .driftDetected(true)
+ .brokerCount(2)
+ .reachableBrokerCount(2)
+ .comparedFields(Arrays.asList("flushDiskType",
"writeQueueNums"))
+ .brokers(Arrays.asList(
+ BrokerConfigDiffVO.BrokerStatusVO.builder()
+ .name("broker-a")
+ .address("10.0.0.1:10911")
+ .reachable(true)
+ .build(),
+ BrokerConfigDiffVO.BrokerStatusVO.builder()
+ .name("broker-b")
+ .address("10.0.0.2:10911")
+ .reachable(true)
+ .build()))
+ .differences(Collections.singletonList(
+ BrokerConfigDiffVO.ConfigDifferenceVO.builder()
+ .field("writeQueueNums")
+
.brokerProperty("defaultTopicQueueNums")
+ .values(Arrays.asList(
+
BrokerConfigDiffVO.ConfigValueVO.builder()
+ .brokerName("broker-a")
+
.address("10.0.0.1:10911")
+ .configured(true)
+ .value("8")
+ .build(),
+
BrokerConfigDiffVO.ConfigValueVO.builder()
+ .brokerName("broker-b")
+
.address("10.0.0.2:10911")
+ .configured(true)
+ .value("16")
+ .build()))
+ .build()))
+ .build());
+
+ mockMvc.perform(get("/api/clusters/cluster-1/broker-config-diff")
+ .param("instanceId", "instance-1"))
+ .andExpect(status().isOk())
+ .andExpect(jsonPath("$.code").value(200))
+ .andExpect(jsonPath("$.data.complete").value(true))
+ .andExpect(jsonPath("$.data.driftDetected").value(true))
+ .andExpect(jsonPath("$.data.brokerCount").value(2))
+ .andExpect(jsonPath("$.data.reachableBrokerCount").value(2))
+
.andExpect(jsonPath("$.data.brokers[0].address").value("10.0.0.1:10911"))
+
.andExpect(jsonPath("$.data.differences[0].field").value("writeQueueNums"))
+
.andExpect(jsonPath("$.data.differences[0].brokerProperty").value("defaultTopicQueueNums"))
+
.andExpect(jsonPath("$.data.differences[0].values[1].value").value("16"));
+
+ verify(brokerConfigDiffService).compare("cluster-1", "instance-1");
+ }
+
@Test
void updateConfigShouldRejectNullRequestBody() throws Exception {
mockMvc.perform(post("/api/clusters/config/update")
diff --git a/web/src/api/cluster.test.ts b/web/src/api/cluster.test.ts
index 21a48aa1f..4a2db972b 100644
--- a/web/src/api/cluster.test.ts
+++ b/web/src/api/cluster.test.ts
@@ -23,6 +23,7 @@ import {
createNameServer,
deleteK8sCert,
deleteNameServer,
+ getBrokerConfigDiff,
getNameServerConfigDiff,
getCluster,
listK8sCerts,
@@ -236,6 +237,43 @@ describe('K8s certificate API', () => {
await expect(getNameServerConfigDiff('cluster-1',
'instance-proxy-1')).resolves.toEqual(result);
});
+ it('loads broker configuration drift for the selected cluster', async () => {
+ const result = {
+ cluster: 'cluster/prod:1',
+ complete: true,
+ driftDetected: true,
+ brokerCount: 2,
+ reachableBrokerCount: 2,
+ comparedFields: ['flushDiskType', 'writeQueueNums'],
+ brokers: [
+ { name: 'broker-a', address: '10.0.0.1:10911', reachable: true },
+ { name: 'broker-b', address: '10.0.0.2:10911', reachable: true },
+ ],
+ differences: [
+ {
+ field: 'writeQueueNums',
+ brokerProperty: 'defaultTopicQueueNums',
+ values: [
+ { brokerName: 'broker-a', address: '10.0.0.1:10911', configured:
true, value: '8' },
+ { brokerName: 'broker-b', address: '10.0.0.2:10911', configured:
true, value: '16' },
+ ],
+ },
+ ],
+ };
+ mock
+ .onGet('/clusters/cluster%2Fprod%3A1/broker-config-diff', {
+ params: { instanceId: 'instance-proxy-1' },
+ })
+ .reply(200, {
+ code: 200,
+ data: result,
+ });
+
+ await expect(getBrokerConfigDiff('cluster/prod:1',
'instance-proxy-1')).resolves.toEqual(
+ result,
+ );
+ });
+
it('sends the proxy restart target', async () => {
const target = { clusterId: 'cluster-1', addr: '127.0.0.1:8081' };
mock.onPost('/proxies/restart').reply((config) => {
diff --git a/web/src/api/cluster.ts b/web/src/api/cluster.ts
index 44c6f9d38..c5a70851d 100644
--- a/web/src/api/cluster.ts
+++ b/web/src/api/cluster.ts
@@ -122,6 +122,37 @@ export interface ClusterConfigPreviewResult {
changed: boolean;
}
+export interface BrokerConfigDiffBroker {
+ name: string;
+ address: string;
+ reachable: boolean;
+ message?: string | null;
+}
+
+export interface BrokerConfigDiffValue {
+ brokerName: string;
+ address: string;
+ configured: boolean;
+ value: string | null;
+}
+
+export interface BrokerConfigDifference {
+ field: string;
+ brokerProperty: string;
+ values: BrokerConfigDiffValue[];
+}
+
+export interface BrokerConfigDiffResult {
+ cluster: string;
+ complete: boolean;
+ driftDetected: boolean;
+ brokerCount: number;
+ reachableBrokerCount: number;
+ comparedFields: string[];
+ brokers: BrokerConfigDiffBroker[];
+ differences: BrokerConfigDifference[];
+}
+
export interface ClusterProbeResult {
connected: boolean;
namesrvAddr: string;
@@ -216,6 +247,14 @@ export async function previewClusterConfig(
return res.data.data;
}
+export async function getBrokerConfigDiff(clusterId: string, instanceId?:
string) {
+ const res = await client.get<{ data: BrokerConfigDiffResult }>(
+ `/clusters/${pathSegment(clusterId)}/broker-config-diff`,
+ { params: instanceId ? { instanceId } : undefined },
+ );
+ return res.data.data;
+}
+
export async function restartBroker(clusterId: string, brokerName: string) {
const res = await client.post<{ data: { success: boolean; message: string }
}>(
`/clusters/${pathSegment(clusterId)}/brokers/${pathSegment(brokerName)}/restart`,
diff --git a/web/src/i18n/translations.ts b/web/src/i18n/translations.ts
index a5dfbc65d..4be81ab23 100644
--- a/web/src/i18n/translations.ts
+++ b/web/src/i18n/translations.ts
@@ -1343,12 +1343,41 @@ const translations: Record<string, Record<Lang,
string>> = {
'cluster.configPreviewProposed': { zh: '预期值', en: 'Proposed Value' },
'cluster.configPreviewProperty': { zh: 'Broker 属性', en: 'Broker Property' },
'cluster.configPreviewNoChanges': { zh: '无配置变更', en: 'No config changes' },
+ 'cluster.brokerConfigDiff': { zh: '配置差异', en: 'Config Diff' },
+ 'cluster.brokerConfigDiffTitle': {
+ zh: 'Broker 配置差异 - {name}',
+ en: 'Broker Config Diff - {name}',
+ },
+ 'cluster.brokerConfigDiffLoading': {
+ zh: '正在检测 Broker 配置差异',
+ en: 'Checking broker config drift',
+ },
+ 'cluster.brokerConfigDiffFailed': {
+ zh: 'Broker 配置差异检测失败',
+ en: 'Failed to check broker config drift',
+ },
+ 'cluster.brokerConfigDiffDriftDetected': {
+ zh: '检测到 Broker 配置不一致',
+ en: 'Broker config drift detected',
+ },
+ 'cluster.brokerConfigDiffNoDrift': {
+ zh: 'Broker 配置一致',
+ en: 'Broker configurations are consistent',
+ },
+ 'cluster.brokerConfigDiffComplete': { zh: '检测完整', en: 'Complete' },
+ 'cluster.brokerConfigDiffComparedFields': { zh: '比较配置项', en: 'Compared
Fields' },
+ 'cluster.brokerConfigDiffValues': { zh: 'Broker 配置值', en: 'Broker Values' },
+ 'cluster.brokerConfigDiffReachable': { zh: '可读', en: 'Readable' },
+ 'cluster.brokerConfigDiffUnreachable': { zh: '不可读', en: 'Unreadable' },
+ 'cluster.brokerConfigDiffUnconfigured': { zh: '未配置', en: 'Unconfigured' },
'cluster.flushDiskType': { zh: '刷盘方式', en: 'Flush Disk Type' },
'cluster.syncFlush': { zh: '同步刷盘', en: 'Sync Flush' },
'cluster.asyncFlush': { zh: '异步刷盘', en: 'Async Flush' },
'cluster.autoCreateTopic': { zh: '自动创建 Topic', en: 'Auto Create Topic' },
'cluster.autoCreateSubGroup': { zh: '自动创建订阅组', en: 'Auto Create Subscription
Group' },
'cluster.maxMessageSize': { zh: '最大消息大小 (MB)', en: 'Max Message Size (MB)' },
+ 'cluster.msgTraceTopicName': { zh: '消息轨迹 Topic', en: 'Message Trace Topic' },
+ 'cluster.deleteWhen': { zh: '删除文件时间', en: 'Delete When' },
'cluster.fileReservedTime': { zh: '文件保留时长 (小时)', en: 'File Retention Time
(hours)' },
'cluster.writeQueues': { zh: '写队列数', en: 'Write Queues' },
'cluster.readQueues': { zh: '读队列数', en: 'Read Queues' },
diff --git a/web/src/pages/cluster/__tests__/ClusterPage.test.tsx
b/web/src/pages/cluster/__tests__/ClusterPage.test.tsx
index 31cf80579..91b354d1e 100644
--- a/web/src/pages/cluster/__tests__/ClusterPage.test.tsx
+++ b/web/src/pages/cluster/__tests__/ClusterPage.test.tsx
@@ -27,6 +27,7 @@ import { LangProvider } from '../../../i18n/LangContext';
const clusterServiceMocks = vi.hoisted(() => ({
createNameserverRegistry: vi.fn(),
deleteNameserverRegistry: vi.fn(),
+ getBrokerConfigDiff: vi.fn(),
getNameServerConfigDiff: vi.fn(),
listClusters: vi.fn(),
listK8sCerts: vi.fn(),
@@ -108,6 +109,15 @@ const buildCluster = ({
tpsIn,
tpsOut,
},
+ {
+ name: 'rocketmq-prod-1',
+ addr: '10.101.2.12:10911',
+ version: '5.2.0',
+ status: 'running',
+ diskUsage: 58,
+ tpsIn: Math.max(tpsIn - 100, 0),
+ tpsOut: Math.max(tpsOut - 100, 0),
+ },
],
proxies: [
{
@@ -183,7 +193,7 @@ describe('Cluster page', () => {
]);
renderWithRoute(<ClusterPage />, '/cluster?instanceId=instance-b');
- await screen.findByText('ns-prod');
+ await screen.findAllByText('ns-prod');
expect(clusterServiceMocks.listClusters).toHaveBeenCalledWith('instance-b');
});
@@ -252,6 +262,19 @@ describe('Cluster page', () => {
nodes: [{ address: 'rocketmq1-nameserver:9876', reachable: true }],
differences: [],
});
+ clusterServiceMocks.getBrokerConfigDiff.mockReset().mockResolvedValue({
+ cluster: 'cluster-prod',
+ complete: true,
+ driftDetected: false,
+ brokerCount: 2,
+ reachableBrokerCount: 2,
+ comparedFields: ['flushDiskType', 'writeQueueNums'],
+ brokers: [
+ { name: 'rocketmq-prod-0', address: '10.101.2.11:10911', reachable:
true },
+ { name: 'rocketmq-prod-1', address: '10.101.2.12:10911', reachable:
true },
+ ],
+ differences: [],
+ });
clusterServiceMocks.listNameserverRegistry.mockReset().mockResolvedValue([
{
id: 1,
@@ -359,7 +382,7 @@ describe('Cluster page', () => {
renderWithProviders(<ClusterPage />);
const brokerRow = await screen.findByRole('row', { name:
/10\.101\.2\.11:10911/ });
- await user.click(within(brokerRow).getByRole('button', { name: /配\s*置/ }));
+ await user.click(within(brokerRow).getByRole('button', { name: /^配\s*置$/
}));
const dialog = await screen.findByRole('dialog', { name: /配置 -
rocketmq-prod/ });
const writeQueuesInput = within(dialog).getByLabelText('写队列数');
await user.clear(writeQueuesInput);
@@ -528,6 +551,84 @@ describe('Cluster page', () => {
).toBeInTheDocument();
});
+ it('opens Broker config drift details from a broker row', async () => {
+ const user = userEvent.setup();
+ clusterServiceMocks.getBrokerConfigDiff.mockResolvedValue({
+ cluster: 'cluster-prod',
+ complete: true,
+ driftDetected: true,
+ brokerCount: 2,
+ reachableBrokerCount: 2,
+ comparedFields: ['flushDiskType', 'writeQueueNums'],
+ brokers: [
+ { name: 'rocketmq-prod-0', address: '10.101.2.11:10911', reachable:
true },
+ { name: 'rocketmq-prod-1', address: '10.101.2.12:10911', reachable:
true },
+ ],
+ differences: [
+ {
+ field: 'writeQueueNums',
+ brokerProperty: 'defaultTopicQueueNums',
+ values: [
+ {
+ brokerName: 'rocketmq-prod-0',
+ address: '10.101.2.11:10911',
+ configured: true,
+ value: '8',
+ },
+ {
+ brokerName: 'rocketmq-prod-1',
+ address: '10.101.2.12:10911',
+ configured: true,
+ value: '16',
+ },
+ ],
+ },
+ ],
+ });
+ renderWithProviders(<ClusterPage />);
+
+ await user.click(screen.getByRole('tab', { name: /Broker 管理/ }));
+ const brokerRow = await screen.findByRole('row', { name:
/10\.101\.2\.11:10911/ });
+ await user.click(within(brokerRow).getByRole('button', { name: /配置差异/ }));
+
+ await waitFor(() =>
+ expect(clusterServiceMocks.getBrokerConfigDiff).toHaveBeenCalledWith(
+ 'cluster-prod',
+ 'instance-1',
+ ),
+ );
+ const dialog = await screen.findByRole('dialog', {
+ name: /Broker 配置差异 - ns-prod/,
+ });
+ expect(within(dialog).getByText('检测到 Broker 配置不一致')).toBeInTheDocument();
+ expect(within(dialog).getByText('2/2')).toBeInTheDocument();
+ expect(within(dialog).getAllByText('写队列数').length).toBeGreaterThan(0);
+
expect(within(dialog).getByText('defaultTopicQueueNums')).toBeInTheDocument();
+ expect(
+ within(dialog).getByText((content) => content.includes('rocketmq-prod-0:
8')),
+ ).toBeInTheDocument();
+ expect(
+ within(dialog).getByText((content) => content.includes('rocketmq-prod-1:
16')),
+ ).toBeInTheDocument();
+ });
+
+ it('reports Broker config drift load failures without closing the dialog',
async () => {
+ const user = userEvent.setup();
+ const errorSpy = vi.spyOn(message, 'error').mockImplementation(vi.fn());
+ clusterServiceMocks.getBrokerConfigDiff.mockRejectedValueOnce(new
Error('failure'));
+ renderWithProviders(<ClusterPage />);
+
+ await user.click(screen.getByRole('tab', { name: /Broker 管理/ }));
+ const brokerRow = await screen.findByRole('row', { name:
/10\.101\.2\.11:10911/ });
+ await user.click(within(brokerRow).getByRole('button', { name: /配置差异/ }));
+
+ await waitFor(() => expect(errorSpy).toHaveBeenCalledWith('Broker
配置差异检测失败'));
+ const dialog = await screen.findByRole('dialog', {
+ name: /Broker 配置差异 - ns-prod/,
+ });
+ expect(within(dialog).getByText('正在检测 Broker 配置差异')).toBeInTheDocument();
+ });
+
it('polls the API after two seconds and renders only returned metrics',
async () => {
vi.useFakeTimers();
const randomSpy = vi.spyOn(Math, 'random');
diff --git a/web/src/pages/cluster/index.tsx b/web/src/pages/cluster/index.tsx
index 78746dbc5..6362ec5d5 100644
--- a/web/src/pages/cluster/index.tsx
+++ b/web/src/pages/cluster/index.tsx
@@ -53,6 +53,8 @@ import PageHeader from '../../components/PageHeader';
import { useLang } from '../../i18n/LangContext';
import { countClusterComponents } from './clusterStats';
import type {
+ BrokerConfigDiffResult,
+ BrokerConfigDifference,
BrokerInfo,
ProxyInfo,
NameserverRegistryEntry,
@@ -67,6 +69,7 @@ import type {
import {
createNameserverRegistry,
deleteNameserverRegistry,
+ getBrokerConfigDiff,
getNameServerConfigDiff,
listClusters,
listK8sCerts,
@@ -93,6 +96,7 @@ type ProxyDetail = ProxyInfo & { clusterId: string;
clusterName: string; nsClust
type ClusterConfigFormValues = Partial<ClusterConfig> & { maxMessageSizeMB:
number };
type ClusterConfigRequest = { id: string; instanceId?: string } &
Partial<ClusterConfig>;
type NameServerConfigDiffNode = NameServerConfigDiffResult['nodes'][number];
+type BrokerConfigDiffBroker = BrokerConfigDiffResult['brokers'][number];
const safeText = (value: string | null | undefined) => value ?? '';
const searchText = (value: string | null | undefined) =>
safeText(value).toLowerCase();
@@ -108,6 +112,8 @@ const CONFIG_FIELD_LABEL_KEYS: Record<string, string> = {
writeQueueNums: 'cluster.writeQueues',
readQueueNums: 'cluster.readQueues',
brokerPermission: 'cluster.brokerPermission',
+ deleteWhen: 'cluster.deleteWhen',
+ msgTraceTopicName: 'cluster.msgTraceTopicName',
};
// ─── Page
─────────────────────────────────────────────────────────────────────
@@ -143,6 +149,17 @@ const ClusterPage = () => {
cluster: null,
result: null,
});
+ const [brokerConfigDiffState, setBrokerConfigDiffState] = useState<{
+ open: boolean;
+ loading: boolean;
+ cluster: ClusterInfo | null;
+ result: BrokerConfigDiffResult | null;
+ }>({
+ open: false,
+ loading: false,
+ cluster: null,
+ result: null,
+ });
const [configForm] = Form.useForm();
const [k8sIdOptions, setK8sIdOptions] = useState<string[]>([]);
@@ -333,6 +350,30 @@ const ClusterPage = () => {
[t],
);
+ const openBrokerConfigDiff = useCallback(
+ async (cluster: ClusterInfo) => {
+ setBrokerConfigDiffState({
+ open: true,
+ loading: true,
+ cluster,
+ result: null,
+ });
+ try {
+ const result = await getBrokerConfigDiff(cluster.id,
selectedInstanceIdRef.current);
+ setBrokerConfigDiffState({
+ open: true,
+ loading: false,
+ cluster,
+ result,
+ });
+ } catch {
+ setBrokerConfigDiffState((current) => ({ ...current, loading: false
}));
+ message.error(t('cluster.brokerConfigDiffFailed'));
+ }
+ },
+ [t],
+ );
+
// ─── Connection test ──────────────────────────────────────────────────────
const [connectModalOpen, setConnectModalOpen] = useState(false);
const [connectTesting, setConnectTesting] = useState(false);
@@ -865,6 +906,154 @@ const ClusterPage = () => {
);
}
+ function renderBrokerConfigDiffModal() {
+ const { cluster, loading: diffLoading, open, result } =
brokerConfigDiffState;
+ const titleName = cluster?.nsClusterName ?? cluster?.name ??
result?.cluster ?? '-';
+ const brokerColumns: ColumnsType<BrokerConfigDiffBroker> = [
+ {
+ title: t('cluster.brokerName'),
+ dataIndex: 'name',
+ key: 'name',
+ width: 180,
+ render: (name: string) => <Text strong>{name}</Text>,
+ },
+ {
+ title: t('common.address'),
+ dataIndex: 'address',
+ key: 'address',
+ render: (address: string) => <Text copyable>{address}</Text>,
+ },
+ {
+ title: t('common.status'),
+ dataIndex: 'reachable',
+ key: 'reachable',
+ width: 120,
+ render: (reachable: boolean) => (
+ <Tag color={reachable ? 'green' : 'red'}>
+ {reachable
+ ? t('cluster.brokerConfigDiffReachable')
+ : t('cluster.brokerConfigDiffUnreachable')}
+ </Tag>
+ ),
+ },
+ {
+ title: t('common.message'),
+ dataIndex: 'message',
+ key: 'message',
+ ellipsis: true,
+ render: (value?: string | null) => value || <Text
type="secondary">-</Text>,
+ },
+ ];
+ const differenceColumns: ColumnsType<BrokerConfigDifference> = [
+ {
+ title: t('cluster.configPreviewField'),
+ dataIndex: 'field',
+ key: 'field',
+ width: 180,
+ render: (field: string) => <Text
strong>{configFieldLabel(field)}</Text>,
+ },
+ {
+ title: t('cluster.configPreviewProperty'),
+ dataIndex: 'brokerProperty',
+ key: 'brokerProperty',
+ width: 190,
+ render: (value: string) => <Text code>{value}</Text>,
+ },
+ {
+ title: t('cluster.brokerConfigDiffValues'),
+ dataIndex: 'values',
+ key: 'values',
+ render: (values: BrokerConfigDifference['values']) => (
+ <Space size={[0, 4]} wrap>
+ {values.map((value) => (
+ <Tag
+ key={`${value.address}-${value.value ?? 'missing'}`}
+ color={value.configured ? 'blue' : 'default'}
+ >
+ {`${value.brokerName || value.address}: ${
+ value.configured
+ ? (value.value ?? '-')
+ : t('cluster.brokerConfigDiffUnconfigured')
+ }`}
+ </Tag>
+ ))}
+ </Space>
+ ),
+ },
+ ];
+
+ return (
+ <Modal
+ title={t('cluster.brokerConfigDiffTitle', { name: titleName })}
+ open={open}
+ onCancel={() =>
+ setBrokerConfigDiffState({ open: false, loading: false, cluster:
null, result: null })
+ }
+ footer={
+ <Button
+ onClick={() =>
+ setBrokerConfigDiffState({ open: false, loading: false, cluster:
null, result: null })
+ }
+ >
+ {t('common.close')}
+ </Button>
+ }
+ width={980}
+ destroyOnHidden
+ >
+ <Spin spinning={diffLoading}>
+ {result ? (
+ <>
+ <Alert
+ showIcon
+ type={result.driftDetected ? 'warning' : 'success'}
+ message={
+ result.driftDetected
+ ? t('cluster.brokerConfigDiffDriftDetected')
+ : t('cluster.brokerConfigDiffNoDrift')
+ }
+ style={{ marginBottom: 16 }}
+ />
+ <Descriptions size="small" column={2} style={{ marginBottom: 16
}}>
+ <Descriptions.Item label={t('cluster.configPreviewTargets')}>
+ {`${result.reachableBrokerCount}/${result.brokerCount}`}
+ </Descriptions.Item>
+ <Descriptions.Item
label={t('cluster.brokerConfigDiffComplete')}>
+ {result.complete ? t('common.yes') : t('common.no')}
+ </Descriptions.Item>
+ <Descriptions.Item
label={t('cluster.brokerConfigDiffComparedFields')} span={2}>
+ <Space size={[0, 4]} wrap>
+ {result.comparedFields.map((field) => (
+ <Tag key={field}>{configFieldLabel(field)}</Tag>
+ ))}
+ </Space>
+ </Descriptions.Item>
+ </Descriptions>
+ <Table<BrokerConfigDiffBroker>
+ columns={brokerColumns}
+ dataSource={result.brokers}
+ rowKey="address"
+ pagination={false}
+ size="small"
+ style={{ marginBottom: 16 }}
+ />
+ <Table<BrokerConfigDifference>
+ columns={differenceColumns}
+ dataSource={result.differences}
+ rowKey="field"
+ pagination={false}
+ size="small"
+ locale={{ emptyText: t('cluster.configPreviewNoChanges') }}
+ />
+ </>
+ ) : (
+ <Alert showIcon type="info"
message={t('cluster.brokerConfigDiffLoading')} />
+ )}
+ </Spin>
+ </Modal>
+ );
+ }
+
function renderBrokerTab() {
type BrokerWithCluster = BrokerInfo & {
clusterName: string;
@@ -986,12 +1175,25 @@ const ClusterPage = () => {
{
title: t('common.actions'),
key: 'action',
- width: 160,
+ width: 260,
render: (_: unknown, record: BrokerWithCluster) => (
<Flex gap={6}>
+ <Button
+ size="small"
+ icon={<EyeOutlined />}
+ aria-label={t('cluster.brokerConfigDiff')}
+ loading={
+ brokerConfigDiffState.loading &&
+ brokerConfigDiffState.cluster?.id === record.cluster.id
+ }
+ onClick={() => void openBrokerConfigDiff(record.cluster)}
+ >
+ {t('cluster.brokerConfigDiff')}
+ </Button>
<Button
size="small"
icon={<SettingOutlined />}
+ aria-label={t('cluster.config')}
style={{ borderColor: '#1677ff', color: '#1677ff' }}
onClick={() => handleConfigOpen(record.cluster)}
>
@@ -1000,6 +1202,7 @@ const ClusterPage = () => {
<Button
size="small"
icon={<ReloadOutlined />}
+ aria-label={t('cluster.restart')}
danger
style={{ borderColor: '#ff4d4f', color: '#ff4d4f' }}
onClick={() => message.warning(t('cluster.restartNotSupported'))}
@@ -1594,6 +1797,7 @@ const ClusterPage = () => {
</Modal>
{renderNameServerConfigDiffModal()}
+ {renderBrokerConfigDiffModal()}
<Modal
title={t('cluster.testConnectionTitle')}
diff --git a/web/src/services/clusterService.ts
b/web/src/services/clusterService.ts
index 8fc5f5c7d..823ef557e 100644
--- a/web/src/services/clusterService.ts
+++ b/web/src/services/clusterService.ts
@@ -1,6 +1,7 @@
import { isMockMode } from './dataMode';
import * as clusterApi from '../api/cluster';
import type {
+ BrokerConfigDiffResult,
ClusterConfig,
ClusterConfigPreviewResult,
ClusterConfigUpdateResult,
@@ -141,6 +142,59 @@ export async function getNameServerConfigDiff(
};
}
+export async function getBrokerConfigDiff(
+ clusterId: string,
+ instanceId?: string,
+): Promise<BrokerConfigDiffResult> {
+ if (!isMockMode()) return clusterApi.getBrokerConfigDiff(clusterId,
instanceId);
+
+ const cluster = getMockCluster(clusterId);
+ const brokers = cluster.brokers
+ .filter((broker) => broker.addr)
+ .map((broker) => ({
+ name: broker.name,
+ address: broker.addr,
+ reachable: String(broker.status) !== 'offline',
+ }));
+ const reachableBrokers = brokers.filter((broker) => broker.reachable);
+ const driftDetected = cluster.id === 'cluster-prod' &&
reachableBrokers.length > 1;
+
+ return {
+ cluster: cluster.id,
+ complete: reachableBrokers.length === brokers.length,
+ driftDetected,
+ brokerCount: brokers.length,
+ reachableBrokerCount: reachableBrokers.length,
+ comparedFields: [
+ 'flushDiskType',
+ 'autoCreateTopicEnable',
+ 'autoCreateSubscriptionGroup',
+ 'maxMessageSize',
+ 'msgTraceTopicName',
+ 'deleteWhen',
+ 'fileReservedTime',
+ 'writeQueueNums',
+ 'readQueueNums',
+ 'brokerPermission',
+ ],
+ brokers,
+ differences: driftDetected
+ ? [
+ {
+ field: 'writeQueueNums',
+ brokerProperty: 'defaultTopicQueueNums',
+ values: reachableBrokers.map((broker, index) => ({
+ brokerName: broker.name,
+ address: broker.address,
+ configured: true,
+ value: index === 0 ? String(cluster.config.writeQueueNums) :
'16',
+ })),
+ },
+ ]
+ : [],
+ };
+}
+
export async function listK8sCerts(): Promise<K8sCertInfo[]> {
if (isMockMode()) {
return mockCertStore.map((cert) => ({ ...cert, san: cert.san ?
[...cert.san] : cert.san }));
