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 1b34fadb fix: keep cluster config updates atomic on repository
failures (#730)
1b34fadb is described below
commit 1b34fadb1017be62ab7f137239baae68c32424e5
Author: Rui <[email protected]>
AuthorDate: Mon Aug 3 11:15:32 2026 +0800
fix: keep cluster config updates atomic on repository failures (#730)
---
.../studio/cluster/broker/ClusterService.java | 25 ++++++++--
.../studio/cluster/broker/ClusterServiceTest.java | 56 ++++++++++++++++++++++
2 files changed, 76 insertions(+), 5 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 5231f358..45393dc1 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
@@ -59,10 +59,7 @@ public class ClusterService {
ClusterVO cluster = clusterRepository.findById(command.getId())
.orElseThrow(() -> new BusinessException(404, "Cluster not
found: " + command.getId()));
- ClusterConfigVO config = cluster.getConfig();
- if (config == null) {
- config = new ClusterConfigVO();
- }
+ ClusterConfigVO config = copyConfig(cluster.getConfig());
if (command.getFlushDiskType() != null) {
config.setFlushDiskType(parseFlushDiskType(command.getFlushDiskType()));
@@ -89,12 +86,30 @@ public class ClusterService {
config.setBrokerPermission(command.getBrokerPermission());
}
- cluster.setConfig(config);
clusterRepository.updateConfig(command.getId(), config);
+ cluster.setConfig(config);
log.info("Cluster config updated successfully for: {}",
command.getId());
return cluster;
}
+ private ClusterConfigVO copyConfig(ClusterConfigVO config) {
+ if (config == null) {
+ return new ClusterConfigVO();
+ }
+ return ClusterConfigVO.builder()
+ .writeQueueNums(config.getWriteQueueNums())
+ .readQueueNums(config.getReadQueueNums())
+ .maxMessageSize(config.getMaxMessageSize())
+ .msgTraceTopicName(config.getMsgTraceTopicName())
+ .autoCreateTopicEnable(config.isAutoCreateTopicEnable())
+
.autoCreateSubscriptionGroup(config.isAutoCreateSubscriptionGroup())
+ .deleteWhen(config.getDeleteWhen())
+ .fileReservedTime(config.getFileReservedTime())
+ .flushDiskType(config.getFlushDiskType())
+ .brokerPermission(config.getBrokerPermission())
+ .build();
+ }
+
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 c5685e33..5cb0b741 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
@@ -47,6 +47,7 @@ import static org.assertj.core.api.Assertions.assertThatCode;
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.doThrow;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -88,8 +89,10 @@ class ClusterServiceTest {
.writeQueueNums(8)
.readQueueNums(8)
.maxMessageSize(4194304)
+ .msgTraceTopicName("RMQ_SYS_TRACE_TOPIC")
.autoCreateTopicEnable(true)
.autoCreateSubscriptionGroup(true)
+ .deleteWhen("04")
.fileReservedTime(72)
.brokerPermission(6)
.build())
@@ -196,6 +199,7 @@ class ClusterServiceTest {
@Test
void updateConfigShouldPreserveExistingValuesWhenCommandFieldsAreNull() {
when(clusterRepository.findById("cluster-1")).thenReturn(Optional.of(sampleCluster));
+ ClusterConfigVO storedConfig = sampleCluster.getConfig();
UpdateConfigDTO command = UpdateConfigDTO.builder()
.id("cluster-1")
@@ -205,10 +209,18 @@ class ClusterServiceTest {
ClusterVO result = clusterService.updateClusterConfig(command);
ClusterConfigVO config = result.getConfig();
+ assertThat(config).isNotSameAs(storedConfig);
assertThat(config.getFlushDiskType()).isEqualTo(FlushDiskType.SYNC_FLUSH);
assertThat(config.getWriteQueueNums()).isEqualTo(8);
assertThat(config.getReadQueueNums()).isEqualTo(8);
+ assertThat(config.getMaxMessageSize()).isEqualTo(4194304);
+
assertThat(config.getMsgTraceTopicName()).isEqualTo("RMQ_SYS_TRACE_TOPIC");
assertThat(config.isAutoCreateTopicEnable()).isTrue();
+ assertThat(config.isAutoCreateSubscriptionGroup()).isTrue();
+ assertThat(config.getDeleteWhen()).isEqualTo("04");
+ assertThat(config.getFileReservedTime()).isEqualTo(72);
+ assertThat(config.getBrokerPermission()).isEqualTo(6);
+
assertThat(storedConfig.getFlushDiskType()).isEqualTo(FlushDiskType.ASYNC_FLUSH);
}
@Test
@@ -263,6 +275,50 @@ class ClusterServiceTest {
assertThat(result.getConfig().getFlushDiskType()).isEqualTo(FlushDiskType.ASYNC_FLUSH);
}
+ @Test
+ void updateConfigShouldNotMutateStoredConfigWhenRepositoryUpdateFails() {
+ ClusterConfigVO storedConfig = sampleCluster.getConfig();
+ RuntimeException persistenceFailure = new
RuntimeException("persistence failed");
+
when(clusterRepository.findById("cluster-1")).thenReturn(Optional.of(sampleCluster));
+ doThrow(persistenceFailure).when(clusterRepository)
+ .updateConfig(eq("cluster-1"), any(ClusterConfigVO.class));
+
+ UpdateConfigDTO command = UpdateConfigDTO.builder()
+ .id("cluster-1")
+ .flushDiskType("SYNC_FLUSH")
+ .writeQueueNums(16)
+ .build();
+
+ assertThatThrownBy(() -> clusterService.updateClusterConfig(command))
+ .isSameAs(persistenceFailure);
+ assertThat(sampleCluster.getConfig()).isSameAs(storedConfig);
+
assertThat(storedConfig.getFlushDiskType()).isEqualTo(FlushDiskType.ASYNC_FLUSH);
+ assertThat(storedConfig.getWriteQueueNums()).isEqualTo(8);
+ }
+
+ @Test
+ void updateConfigShouldLeaveNullStoredConfigWhenRepositoryUpdateFails() {
+ ClusterVO clusterWithNullConfig = ClusterVO.builder()
+ .name("null-config-cluster")
+ .status(ClusterStatus.healthy)
+ .config(null)
+ .build();
+ clusterWithNullConfig.setId("cluster-nc");
+ RuntimeException persistenceFailure = new
RuntimeException("persistence failed");
+
when(clusterRepository.findById("cluster-nc")).thenReturn(Optional.of(clusterWithNullConfig));
+ doThrow(persistenceFailure).when(clusterRepository)
+ .updateConfig(eq("cluster-nc"), any(ClusterConfigVO.class));
+
+ UpdateConfigDTO command = UpdateConfigDTO.builder()
+ .id("cluster-nc")
+ .flushDiskType("ASYNC_FLUSH")
+ .build();
+
+ assertThatThrownBy(() -> clusterService.updateClusterConfig(command))
+ .isSameAs(persistenceFailure);
+ assertThat(clusterWithNullConfig.getConfig()).isNull();
+ }
+
@Test
void restartBrokerShouldReturnTrue() {
when(clusterRepository.findById("cluster-1")).thenReturn(Optional.of(sampleCluster));
