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 b8f33777 fix: report partial broker configuration updates (#1001)
b8f33777 is described below
commit b8f3377774bec72895d0fccb38a5de1c205c1083
Author: 0 <[email protected]>
AuthorDate: Wed Aug 5 17:43:10 2026 +0800
fix: report partial broker configuration updates (#1001)
Co-authored-by: btlqql <[email protected]>
---
.../studio/cluster/broker/ClusterController.java | 3 +-
.../studio/cluster/broker/ClusterService.java | 78 ++++++++++++++++++----
.../config/BrokerConfigUpdateFailureVO.java | 27 ++++++++
.../config/ClusterConfigUpdateResultVO.java | 38 +++++++++++
.../cluster/broker/ClusterControllerTest.java | 31 +++++++--
.../studio/cluster/broker/ClusterServiceTest.java | 68 +++++++++++++++----
web/src/api/cluster.test.ts | 15 +++++
web/src/api/cluster.ts | 20 +++++-
web/src/i18n/translations.ts | 8 +++
.../pages/cluster/__tests__/ClusterPage.test.tsx | 10 ++-
web/src/pages/cluster/index.tsx | 23 +++++--
web/src/services/clusterService.ts | 22 ++++--
12 files changed, 300 insertions(+), 43 deletions(-)
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 36bd36d1..7192c6c2 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.ClusterConfigUpdateResultVO;
import org.apache.rocketmq.studio.cluster.config.UpdateConfigDTO;
import org.apache.rocketmq.studio.common.domain.Result;
@@ -56,7 +57,7 @@ public class ClusterController {
}
@PostMapping("/config/update")
- public Result<ClusterVO> updateClusterConfig(@Valid @RequestBody(required
= false) UpdateConfigDTO command) {
+ public Result<ClusterConfigUpdateResultVO> updateClusterConfig(@Valid
@RequestBody(required = false) UpdateConfigDTO command) {
requireUpdateConfigCommand(command);
return Result.ok(clusterService.updateClusterConfig(command));
}
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 e57b5cf8..16a8e155 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
@@ -16,6 +16,8 @@
*/
package org.apache.rocketmq.studio.cluster.broker;
+import org.apache.rocketmq.studio.cluster.config.BrokerConfigUpdateFailureVO;
+import org.apache.rocketmq.studio.cluster.config.ClusterConfigUpdateResultVO;
import org.apache.rocketmq.studio.cluster.config.ClusterConfigVO;
import org.apache.rocketmq.studio.cluster.config.UpdateConfigDTO;
import org.apache.rocketmq.studio.cluster.nameserver.CreateNameServerDTO;
@@ -28,11 +30,13 @@ import
org.apache.rocketmq.studio.cluster.proxy.RestartProxyDTO;
import org.apache.rocketmq.studio.common.domain.enums.FlushDiskType;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.ops.audit.AuditService;
import org.apache.rocketmq.studio.rocketmq.RocketMQBrokerConfigService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
+import java.util.ArrayList;
import java.util.List;
import java.util.Properties;
@@ -44,6 +48,7 @@ public class ClusterService {
private final ClusterRepository clusterRepository;
private final ClusterProvider clusterProvider;
private final RocketMQBrokerConfigService brokerConfigService;
+ private final AuditService auditService;
public List<ClusterVO> listClusters() {
log.info("Listing all clusters");
@@ -89,13 +94,51 @@ public class ClusterService {
}
}
- public ClusterVO updateClusterConfig(UpdateConfigDTO command) {
+ public ClusterConfigUpdateResultVO updateClusterConfig(UpdateConfigDTO
command) {
log.info("Updating cluster config for: {}", command.getId());
requireMatchingDefaultQueueNums(command);
ClusterVO cluster = resolveCluster(command.getId());
ClusterConfigVO config = copyConfig(cluster.getConfig());
+ applyConfig(command, config);
+ List<String> successfulBrokers = new ArrayList<>();
+ List<BrokerConfigUpdateFailureVO> failedBrokers = new ArrayList<>();
+ if (cluster.getBrokers() != null && !cluster.getBrokers().isEmpty()) {
+ Properties brokerProps = buildBrokerProperties(command);
+ for (BrokerVO broker : cluster.getBrokers()) {
+ String address = broker.getAddr();
+ if (address == null || address.isEmpty()) {
+ continue;
+ }
+ try {
+ brokerConfigService.updateBrokerConfig(address,
command.getId(), brokerProps);
+ successfulBrokers.add(address);
+ } catch (Exception e) {
+ failedBrokers.add(BrokerConfigUpdateFailureVO.builder()
+ .address(address)
+ .message(e.getMessage())
+ .build());
+ }
+ }
+ }
+
+ ClusterConfigUpdateResultVO.Status status =
updateStatus(successfulBrokers, failedBrokers);
+ if (failedBrokers.isEmpty()) {
+ clusterRepository.updateConfig(command.getId(), config);
+ cluster.setConfig(config);
+ }
+ recordConfigUpdateAudit(command.getId(), status, successfulBrokers,
failedBrokers);
+ log.info("Cluster config update finished for {} with status {}",
command.getId(), status);
+ return ClusterConfigUpdateResultVO.builder()
+ .cluster(cluster)
+ .status(status)
+ .successfulBrokers(List.copyOf(successfulBrokers))
+ .failedBrokers(List.copyOf(failedBrokers))
+ .build();
+ }
+
+ private void applyConfig(UpdateConfigDTO command, ClusterConfigVO config) {
if (command.getFlushDiskType() != null) {
config.setFlushDiskType(parseFlushDiskType(command.getFlushDiskType()));
}
@@ -120,21 +163,30 @@ public class ClusterService {
if (command.getBrokerPermission() != null) {
config.setBrokerPermission(command.getBrokerPermission());
}
+ }
- // Push config to live brokers via admin API
- if (cluster.getBrokers() != null && !cluster.getBrokers().isEmpty()) {
- Properties brokerProps = buildBrokerProperties(command);
- for (BrokerVO broker : cluster.getBrokers()) {
- if (broker.getAddr() != null && !broker.getAddr().isEmpty()) {
- brokerConfigService.updateBrokerConfig(broker.getAddr(),
command.getId(), brokerProps);
- }
- }
+ private ClusterConfigUpdateResultVO.Status updateStatus(
+ List<String> successfulBrokers,
+ List<BrokerConfigUpdateFailureVO> failedBrokers) {
+ if (failedBrokers.isEmpty()) {
+ return ClusterConfigUpdateResultVO.Status.SUCCESS;
}
+ if (successfulBrokers.isEmpty()) {
+ return ClusterConfigUpdateResultVO.Status.FAILED;
+ }
+ return ClusterConfigUpdateResultVO.Status.PARTIAL;
+ }
- clusterRepository.updateConfig(command.getId(), config);
- cluster.setConfig(config);
- log.info("Cluster config updated successfully for: {}",
command.getId());
- return cluster;
+ private void recordConfigUpdateAudit(
+ String clusterId,
+ ClusterConfigUpdateResultVO.Status status,
+ List<String> successfulBrokers,
+ List<BrokerConfigUpdateFailureVO> failedBrokers) {
+ String detail = "successfulBrokers=" + successfulBrokers
+ + ", failedBrokers=" + failedBrokers.stream()
+ .map(failure -> failure.getAddress() + ": " +
failure.getMessage())
+ .toList();
+ auditService.record("UPDATE_CLUSTER_CONFIG", "CLUSTER:" + clusterId,
detail, status.name());
}
private ClusterVO resolveCluster(String clusterId) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/config/BrokerConfigUpdateFailureVO.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/config/BrokerConfigUpdateFailureVO.java
new file mode 100644
index 00000000..5b9d62a6
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/config/BrokerConfigUpdateFailureVO.java
@@ -0,0 +1,27 @@
+/*
+ * 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.Builder;
+import lombok.Value;
+
+@Value
+@Builder
+public class BrokerConfigUpdateFailureVO {
+ String address;
+ String message;
+}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/config/ClusterConfigUpdateResultVO.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/config/ClusterConfigUpdateResultVO.java
new file mode 100644
index 00000000..4fabf6b7
--- /dev/null
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/config/ClusterConfigUpdateResultVO.java
@@ -0,0 +1,38 @@
+/*
+ * 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 org.apache.rocketmq.studio.cluster.broker.ClusterVO;
+import lombok.Builder;
+import lombok.Value;
+
+import java.util.List;
+
+@Value
+@Builder
+public class ClusterConfigUpdateResultVO {
+ ClusterVO cluster;
+ Status status;
+ List<String> successfulBrokers;
+ List<BrokerConfigUpdateFailureVO> failedBrokers;
+
+ public enum Status {
+ SUCCESS,
+ PARTIAL,
+ FAILED
+ }
+}
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 04c438df..b71527b5 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.ClusterConfigUpdateResultVO;
import org.apache.rocketmq.studio.cluster.config.ClusterConfigVO;
import org.apache.rocketmq.studio.cluster.config.UpdateConfigDTO;
@@ -167,7 +168,13 @@ class ClusterControllerTest {
.writeQueueNums(16)
.readQueueNums(16)
.build());
-
when(clusterService.updateClusterConfig(any(UpdateConfigDTO.class))).thenReturn(updated);
+
when(clusterService.updateClusterConfig(any(UpdateConfigDTO.class))).thenReturn(
+ ClusterConfigUpdateResultVO.builder()
+ .cluster(updated)
+ .status(ClusterConfigUpdateResultVO.Status.SUCCESS)
+ .successfulBrokers(Collections.emptyList())
+ .failedBrokers(Collections.emptyList())
+ .build());
UpdateConfigDTO command = UpdateConfigDTO.builder()
.id("cluster-1")
@@ -181,10 +188,11 @@ class ClusterControllerTest {
.content(objectMapper.writeValueAsString(command)))
.andExpect(status().isOk())
.andExpect(jsonPath("$.code").value(200))
- .andExpect(jsonPath("$.data.id").value("cluster-1"))
-
.andExpect(jsonPath("$.data.config.flushDiskType").value("SYNC_FLUSH"))
- .andExpect(jsonPath("$.data.config.writeQueueNums").value(16))
- .andExpect(jsonPath("$.data.config.readQueueNums").value(16));
+ .andExpect(jsonPath("$.data.status").value("SUCCESS"))
+ .andExpect(jsonPath("$.data.cluster.id").value("cluster-1"))
+
.andExpect(jsonPath("$.data.cluster.config.flushDiskType").value("SYNC_FLUSH"))
+
.andExpect(jsonPath("$.data.cluster.config.writeQueueNums").value(16))
+
.andExpect(jsonPath("$.data.cluster.config.readQueueNums").value(16));
}
@Test
@@ -255,7 +263,7 @@ class ClusterControllerTest {
@MethodSource("boundaryConfigValues")
void updateConfigShouldAcceptBoundaryValues(String field, int value)
throws Exception {
when(clusterService.updateClusterConfig(any(UpdateConfigDTO.class)))
- .thenReturn(buildCluster("cluster-1", "production-cluster",
ClusterStatus.healthy));
+ .thenReturn(successfulUpdateResult());
ObjectNode command = objectMapper.createObjectNode()
.put("id", "cluster-1")
.put(field, value);
@@ -272,7 +280,7 @@ class ClusterControllerTest {
@Test
void updateConfigShouldAcceptIdOnlyPartialRequest() throws Exception {
when(clusterService.updateClusterConfig(any(UpdateConfigDTO.class)))
- .thenReturn(buildCluster("cluster-1", "production-cluster",
ClusterStatus.healthy));
+ .thenReturn(successfulUpdateResult());
ObjectNode command = objectMapper.createObjectNode().put("id",
"cluster-1");
mockMvc.perform(post("/api/clusters/config/update")
@@ -295,6 +303,15 @@ class ClusterControllerTest {
.andExpect(jsonPath("$.data.message").value("Broker restart
initiated for broker-0"));
}
+ private ClusterConfigUpdateResultVO successfulUpdateResult() {
+ return ClusterConfigUpdateResultVO.builder()
+ .cluster(buildCluster("cluster-1", "production-cluster",
ClusterStatus.healthy))
+ .status(ClusterConfigUpdateResultVO.Status.SUCCESS)
+ .successfulBrokers(Collections.emptyList())
+ .failedBrokers(Collections.emptyList())
+ .build();
+ }
+
private ClusterVO buildCluster(String id, String name, ClusterStatus
status) {
ClusterVO cluster = ClusterVO.builder()
.name(name)
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 f980f8f0..f9ecfe66 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
@@ -16,6 +16,7 @@
*/
package org.apache.rocketmq.studio.cluster.broker;
+import org.apache.rocketmq.studio.cluster.config.ClusterConfigUpdateResultVO;
import org.apache.rocketmq.studio.cluster.config.ClusterConfigVO;
import org.apache.rocketmq.studio.cluster.config.UpdateConfigDTO;
import org.apache.rocketmq.studio.cluster.nameserver.CreateNameServerDTO;
@@ -31,6 +32,8 @@ import
org.apache.rocketmq.studio.common.domain.enums.ClusterStatus;
import org.apache.rocketmq.studio.common.domain.enums.ClusterType;
import org.apache.rocketmq.studio.common.domain.enums.FlushDiskType;
import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.ops.audit.AuditService;
+import org.apache.rocketmq.studio.rocketmq.RocketMQBrokerConfigService;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
@@ -48,6 +51,7 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
@@ -63,6 +67,12 @@ class ClusterServiceTest {
@Mock
private ClusterProvider clusterProvider;
+ @Mock
+ private RocketMQBrokerConfigService brokerConfigService;
+
+ @Mock
+ private AuditService auditService;
+
@InjectMocks
private ClusterService clusterService;
@@ -178,9 +188,9 @@ class ClusterServiceTest {
.flushDiskType("SYNC_FLUSH")
.build();
- ClusterVO result = clusterService.updateClusterConfig(command);
+ ClusterConfigUpdateResultVO result =
clusterService.updateClusterConfig(command);
-
assertThat(result.getConfig().getFlushDiskType()).isEqualTo(FlushDiskType.SYNC_FLUSH);
+
assertThat(result.getCluster().getConfig().getFlushDiskType()).isEqualTo(FlushDiskType.SYNC_FLUSH);
verify(clusterRepository).updateConfig(eq("cluster-1"),
any(ClusterConfigVO.class));
}
@@ -200,9 +210,9 @@ class ClusterServiceTest {
.brokerPermission(4)
.build();
- ClusterVO result = clusterService.updateClusterConfig(command);
+ ClusterConfigUpdateResultVO result =
clusterService.updateClusterConfig(command);
- ClusterConfigVO config = result.getConfig();
+ ClusterConfigVO config = result.getCluster().getConfig();
assertThat(config.getFlushDiskType()).isEqualTo(FlushDiskType.SYNC_FLUSH);
assertThat(config.isAutoCreateTopicEnable()).isFalse();
assertThat(config.isAutoCreateSubscriptionGroup()).isFalse();
@@ -223,9 +233,9 @@ class ClusterServiceTest {
.flushDiskType("SYNC_FLUSH")
.build();
- ClusterVO result = clusterService.updateClusterConfig(command);
+ ClusterConfigUpdateResultVO result =
clusterService.updateClusterConfig(command);
- ClusterConfigVO config = result.getConfig();
+ ClusterConfigVO config = result.getCluster().getConfig();
assertThat(config).isNotSameAs(storedConfig);
assertThat(config.getFlushDiskType()).isEqualTo(FlushDiskType.SYNC_FLUSH);
assertThat(config.getWriteQueueNums()).isEqualTo(8);
@@ -240,6 +250,39 @@ class ClusterServiceTest {
assertThat(storedConfig.getFlushDiskType()).isEqualTo(FlushDiskType.ASYNC_FLUSH);
}
+ @Test
+ void updateConfigShouldReportPartialFailureAfterOneBrokerSucceeds() {
+ sampleCluster.setBrokers(List.of(
+
BrokerVO.builder().name("broker-0").addr("10.0.0.1:10911").build(),
+
BrokerVO.builder().name("broker-1").addr("10.0.0.2:10911").build()));
+
when(clusterRepository.findById("cluster-1")).thenReturn(Optional.of(sampleCluster));
+ doNothing().when(brokerConfigService).updateBrokerConfig(
+ eq("10.0.0.1:10911"), eq("cluster-1"), any());
+ doThrow(new BusinessException(500, "broker unavailable"))
+ .when(brokerConfigService).updateBrokerConfig(
+ eq("10.0.0.2:10911"), eq("cluster-1"), any());
+
+ ClusterConfigUpdateResultVO result =
clusterService.updateClusterConfig(
+
UpdateConfigDTO.builder().id("cluster-1").writeQueueNums(16).build());
+
+
assertThat(result.getStatus()).isEqualTo(ClusterConfigUpdateResultVO.Status.PARTIAL);
+
assertThat(result.getSuccessfulBrokers()).containsExactly("10.0.0.1:10911");
+
assertThat(result.getFailedBrokers()).singleElement().satisfies(failure -> {
+ assertThat(failure.getAddress()).isEqualTo("10.0.0.2:10911");
+ assertThat(failure.getMessage()).contains("broker unavailable");
+ });
+ verify(brokerConfigService).updateBrokerConfig(
+ eq("10.0.0.1:10911"), eq("cluster-1"), any());
+ verify(brokerConfigService).updateBrokerConfig(
+ eq("10.0.0.2:10911"), eq("cluster-1"), any());
+ verify(clusterRepository, never()).updateConfig(eq("cluster-1"),
any());
+ verify(auditService).record(
+ eq("UPDATE_CLUSTER_CONFIG"),
+ eq("CLUSTER:cluster-1"),
+ org.mockito.ArgumentMatchers.contains("10.0.0.2:10911"),
+ eq("PARTIAL"));
+ }
+
@Test
void updateConfigShouldThrowWhenClusterNotFound() {
when(clusterRepository.findById("missing")).thenReturn(Optional.empty());
@@ -286,10 +329,10 @@ class ClusterServiceTest {
.flushDiskType("ASYNC_FLUSH")
.build();
- ClusterVO result = clusterService.updateClusterConfig(command);
+ ClusterConfigUpdateResultVO result =
clusterService.updateClusterConfig(command);
- assertThat(result.getConfig()).isNotNull();
-
assertThat(result.getConfig().getFlushDiskType()).isEqualTo(FlushDiskType.ASYNC_FLUSH);
+ assertThat(result.getCluster().getConfig()).isNotNull();
+
assertThat(result.getCluster().getConfig().getFlushDiskType()).isEqualTo(FlushDiskType.ASYNC_FLUSH);
}
@Test
@@ -523,10 +566,11 @@ class ClusterServiceTest {
.maxMessageSize(8_388_608)
.build();
- ClusterVO result = clusterService.updateClusterConfig(command);
+ ClusterConfigUpdateResultVO result =
clusterService.updateClusterConfig(command);
-
assertThat(result.getConfig().getMaxMessageSize()).isEqualTo(8_388_608);
- verify(clusterRepository).updateConfig("cluster-1",
result.getConfig());
+
assertThat(result.getCluster().getConfig().getMaxMessageSize()).isEqualTo(8_388_608);
+
assertThat(result.getStatus()).isEqualTo(ClusterConfigUpdateResultVO.Status.SUCCESS);
+ verify(clusterRepository).updateConfig("cluster-1",
result.getCluster().getConfig());
}
private void assertUnsupportedOperation(ThrowableAssert.ThrowingCallable
callable, String message) {
diff --git a/web/src/api/cluster.test.ts b/web/src/api/cluster.test.ts
index 65eea44e..c77a6645 100644
--- a/web/src/api/cluster.test.ts
+++ b/web/src/api/cluster.test.ts
@@ -29,6 +29,7 @@ import {
restartBroker,
restartNameServer,
restartProxy,
+ updateClusterConfig,
updateK8sCert,
updateNameServer,
upgradeNameServer,
@@ -132,6 +133,20 @@ describe('K8s certificate API', () => {
});
});
+ it('returns per-broker cluster config update results', async () => {
+ const result = {
+ cluster: { id: 'cluster-1' },
+ status: 'PARTIAL',
+ successfulBrokers: ['10.0.0.1:10911'],
+ failedBrokers: [{ address: '10.0.0.2:10911', message: 'broker
unavailable' }],
+ };
+ mock.onPost('/clusters/config/update').reply(200, { code: 200, data:
result });
+
+ await expect(updateClusterConfig({ id: 'cluster-1', writeQueueNums: 16
})).resolves.toEqual(
+ result,
+ );
+ });
+
it('sends NameServer operation payloads to their endpoints', async () => {
const target = { clusterId: 'cluster-1', addr: '127.0.0.1:9876' };
const requests = [
diff --git a/web/src/api/cluster.ts b/web/src/api/cluster.ts
index cb8f4ea4..fc0e4a0c 100644
--- a/web/src/api/cluster.ts
+++ b/web/src/api/cluster.ts
@@ -73,6 +73,20 @@ export interface ClusterConfig {
deleteWhen: string;
}
+export type ClusterConfigUpdateStatus = 'SUCCESS' | 'PARTIAL' | 'FAILED';
+
+export interface BrokerConfigUpdateFailure {
+ address: string;
+ message: string;
+}
+
+export interface ClusterConfigUpdateResult {
+ cluster: ClusterInfo;
+ status: ClusterConfigUpdateStatus;
+ successfulBrokers: string[];
+ failedBrokers: BrokerConfigUpdateFailure[];
+}
+
export interface ClusterProbeResult {
connected: boolean;
namesrvAddr: string;
@@ -116,7 +130,11 @@ export async function getCluster(id: string) {
}
export async function updateClusterConfig(data: { id: string } &
Partial<ClusterConfig>) {
- await client.post('/clusters/config/update', data);
+ const res = await client.post<{ data: ClusterConfigUpdateResult }>(
+ '/clusters/config/update',
+ data,
+ );
+ return res.data.data;
}
export async function restartBroker(clusterId: string, brokerName: string) {
diff --git a/web/src/i18n/translations.ts b/web/src/i18n/translations.ts
index a6081e62..d15ccb37 100644
--- a/web/src/i18n/translations.ts
+++ b/web/src/i18n/translations.ts
@@ -546,6 +546,14 @@ const translations: Record<string, Record<Lang, string>> =
{
'cluster.probeElapsed': { zh: '耗时 (ms)', en: 'Elapsed (ms)' },
'cluster.configTitle': { zh: '配置 - {name}', en: 'Config - {name}' },
'cluster.configUpdated': { zh: '配置已更新', en: 'Configuration updated' },
+ 'cluster.configPartiallyUpdated': {
+ zh: '部分 Broker 配置已更新，请检查：{brokers}',
+ en: 'Some broker configurations were updated. Check: {brokers}',
+ },
+ 'cluster.configUpdateFailed': {
+ zh: 'Broker 配置更新失败，请检查：{brokers}',
+ en: 'Broker configuration update failed. Check: {brokers}',
+ },
'cluster.flushDiskType': { zh: '刷盘方式', en: 'Flush Disk Type' },
'cluster.syncFlush': { zh: '同步刷盘', en: 'Sync Flush' },
'cluster.asyncFlush': { zh: '异步刷盘', en: 'Async Flush' },
diff --git a/web/src/pages/cluster/__tests__/ClusterPage.test.tsx
b/web/src/pages/cluster/__tests__/ClusterPage.test.tsx
index b436e577..ca117fd8 100644
--- a/web/src/pages/cluster/__tests__/ClusterPage.test.tsx
+++ b/web/src/pages/cluster/__tests__/ClusterPage.test.tsx
@@ -136,7 +136,15 @@ describe('Cluster page', () => {
clusterServiceMocks.listClusters.mockReset().mockResolvedValue([buildCluster()]);
clusterServiceMocks.restartProxy.mockReset().mockResolvedValue(undefined);
clusterServiceMocks.testClusterConnection.mockReset();
-
clusterServiceMocks.updateClusterConfig.mockReset().mockResolvedValue(undefined);
+
clusterServiceMocks.updateClusterConfig.mockReset().mockImplementation(async ()
=> {
+ const cluster = buildCluster();
+ return {
+ cluster,
+ status: 'SUCCESS',
+ successfulBrokers: cluster.brokers.map((broker) => broker.addr),
+ failedBrokers: [],
+ };
+ });
clusterServiceMocks.updateNameServer.mockReset().mockResolvedValue(undefined);
});
diff --git a/web/src/pages/cluster/index.tsx b/web/src/pages/cluster/index.tsx
index f285d2ee..cab68134 100644
--- a/web/src/pages/cluster/index.tsx
+++ b/web/src/pages/cluster/index.tsx
@@ -526,13 +526,28 @@ const ClusterPage = () => {
...configValues,
maxMessageSize: maxMessageSizeMB * 1048576,
};
- await updateClusterConfig({
+ const result = await updateClusterConfig({
id: selectedCluster.id,
...nextConfig,
});
- await requestRefresh('operation');
- message.success(t('cluster.configUpdated'));
- setConfigModalOpen(false);
+ if (result.status === 'SUCCESS') {
+ await requestRefresh('operation');
+ message.success(t('cluster.configUpdated'));
+ setConfigModalOpen(false);
+ return;
+ }
+
+ const failedAddresses = result.failedBrokers
+ .map((failure) => failure.address)
+ .join(', ');
+ if (result.status === 'PARTIAL') {
+ await requestRefresh('operation');
+ message.warning(
+ t('cluster.configPartiallyUpdated', { brokers:
failedAddresses }),
+ );
+ return;
+ }
+ message.error(t('cluster.configUpdateFailed', { brokers:
failedAddresses }));
});
}}
width={560}
diff --git a/web/src/services/clusterService.ts
b/web/src/services/clusterService.ts
index 601f9f22..6d83bcdc 100644
--- a/web/src/services/clusterService.ts
+++ b/web/src/services/clusterService.ts
@@ -1,6 +1,12 @@
import { isMockMode } from './dataMode';
import * as clusterApi from '../api/cluster';
-import type { ClusterConfig, ClusterInfo, ClusterProbeResult, K8sCertInfo }
from '../api/cluster';
+import type {
+ ClusterConfig,
+ ClusterConfigUpdateResult,
+ ClusterInfo,
+ ClusterProbeResult,
+ K8sCertInfo,
+} from '../api/cluster';
import clusters, { mockK8sCerts } from '../mock/clusters';
const mockCertStore: K8sCertInfo[] = mockK8sCerts.map((cert) => ({
@@ -131,8 +137,14 @@ export async function deleteK8sCert(id: string):
Promise<void> {
export async function updateClusterConfig(data: { id: string } &
Partial<ClusterConfig>) {
if (isMockMode()) {
const { id, ...config } = data;
- Object.assign(getMockCluster(id).config, config);
- return;
+ const cluster = getMockCluster(id);
+ Object.assign(cluster.config, config);
+ return {
+ cluster: copyCluster(cluster),
+ status: 'SUCCESS',
+ successfulBrokers: cluster.brokers.map((broker) => broker.addr),
+ failedBrokers: [],
+ } satisfies ClusterConfigUpdateResult;
}
return clusterApi.updateClusterConfig(data);
}
@@ -208,7 +220,9 @@ export async function updateNameServer(data: {
const nameServer = nameServers.find((item) => item.addr === data.addr);
if (!nameServer) throw new Error(`NameServer not found: ${data.addr}`);
if (data.newAddr && data.newAddr !== data.addr) {
- const duplicate = nameServers.some((item) => item !== nameServer &&
item.addr === data.newAddr);
+ const duplicate = nameServers.some(
+ (item) => item !== nameServer && item.addr === data.newAddr,
+ );
if (duplicate) throw new Error(`NameServer already exists:
${data.newAddr}`);
}
if (data.newAddr) nameServer.addr = data.newAddr;
