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 eec263d7 fix: isolate control-plane audit failures (#1135)
eec263d7 is described below
commit eec263d72f99f2a9c408d09be7391b3ffbce05fc
Author: aias00 <[email protected]>
AuthorDate: Fri Aug 7 00:21:18 2026 -0700
fix: isolate control-plane audit failures (#1135)
* fix: isolate metadata audit failures
* fix: isolate broker config audit failures
---
.../studio/cluster/broker/ClusterService.java | 6 +-
.../provider/apache/RocketMQAdminClientImpl.java | 53 +++++++++-------
.../apache/RocketMQBrokerConfigService.java | 13 +++-
.../studio/cluster/broker/ClusterServiceTest.java | 15 +++++
.../apache/RocketMQAdminClientImplTest.java | 46 ++++++++++++++
.../apache/RocketMQBrokerConfigServiceTest.java | 73 ++++++++++++++++++++++
6 files changed, 179 insertions(+), 27 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 4b8c76ae..8d85d49e 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
@@ -189,7 +189,11 @@ public class ClusterService {
+ ", failedBrokers=" + failedBrokers.stream()
.map(failure -> failure.getAddress() + ": " +
failure.getMessage())
.toList();
- auditService.record("UPDATE_CLUSTER_CONFIG", "CLUSTER:" + clusterId,
detail, status.name());
+ try {
+ auditService.record("UPDATE_CLUSTER_CONFIG", "CLUSTER:" +
clusterId, detail, status.name());
+ } catch (Exception e) {
+ log.warn("Failed to record cluster config update audit for {}:
{}", clusterId, e.getMessage());
+ }
}
private ClusterVO resolveCluster(String clusterId) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
index f6fc162c..34939d5b 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
@@ -193,7 +193,7 @@ public class RocketMQAdminClientImpl implements AdminClient
{
topicMapper.updateById(entity);
}
- auditService.record("CREATE_TOPIC", topicName,
+ recordAudit("CREATE_TOPIC", topicName,
"queues=" + writeQueues + "/" + readQueues, "SUCCESS");
topic.setId(topicName);
@@ -201,10 +201,10 @@ public class RocketMQAdminClientImpl implements
AdminClient {
topic.setReadQueues(readQueues);
return topic;
} catch (BusinessException e) {
- auditService.record("CREATE_TOPIC", topicName, e.getMessage(),
"FAILED");
+ recordAudit("CREATE_TOPIC", topicName, e.getMessage(),
"FAILED");
throw e;
} catch (Exception e) {
- auditService.record("CREATE_TOPIC", topicName, e.getMessage(),
"FAILED");
+ recordAudit("CREATE_TOPIC", topicName, e.getMessage(),
"FAILED");
throw new BusinessException(500, "Failed to create topic: " +
e.getMessage());
}
});
@@ -252,7 +252,7 @@ public class RocketMQAdminClientImpl implements AdminClient
{
topicMapper.updateById(existing);
}
- auditService.record("UPDATE_TOPIC", topicName,
+ recordAudit("UPDATE_TOPIC", topicName,
"queues=" + writeQueues + "/" + readQueues, "SUCCESS");
topic.setId(topicName);
@@ -260,10 +260,10 @@ public class RocketMQAdminClientImpl implements
AdminClient {
topic.setReadQueues(readQueues);
return topic;
} catch (BusinessException e) {
- auditService.record("UPDATE_TOPIC", topicName, e.getMessage(),
"FAILED");
+ recordAudit("UPDATE_TOPIC", topicName, e.getMessage(),
"FAILED");
throw e;
} catch (Exception e) {
- auditService.record("UPDATE_TOPIC", topicName, e.getMessage(),
"FAILED");
+ recordAudit("UPDATE_TOPIC", topicName, e.getMessage(),
"FAILED");
throw new BusinessException(500, "Failed to update topic: " +
e.getMessage());
}
});
@@ -294,13 +294,13 @@ public class RocketMQAdminClientImpl implements
AdminClient {
// Delete from DB
topicMapper.delete(new
LambdaQueryWrapper<RmqTopic>().eq(RmqTopic::getName, name));
- auditService.record("DELETE_TOPIC", name, "", "SUCCESS");
+ recordAudit("DELETE_TOPIC", name, "", "SUCCESS");
return null;
} catch (BusinessException e) {
- auditService.record("DELETE_TOPIC", name, e.getMessage(),
"FAILED");
+ recordAudit("DELETE_TOPIC", name, e.getMessage(), "FAILED");
throw e;
} catch (Exception e) {
- auditService.record("DELETE_TOPIC", name, e.getMessage(),
"FAILED");
+ recordAudit("DELETE_TOPIC", name, e.getMessage(), "FAILED");
throw new BusinessException(500, "Failed to delete topic: " +
e.getMessage());
}
});
@@ -336,12 +336,8 @@ public class RocketMQAdminClientImpl implements
AdminClient {
// The message is already delivered by now; an audit write failure
must not turn a
// successful send into an error, or callers would retry and
duplicate the message.
- try {
- auditService.record("SEND_MESSAGE", topic,
- "tag=" + tag + ", key=" + key + ", msgId=" +
sendResult.getMsgId(), "SUCCESS");
- } catch (Exception auditFailure) {
- log.warn("Failed to record send message audit: {}",
auditFailure.getMessage());
- }
+ recordAudit("SEND_MESSAGE", topic,
+ "tag=" + tag + ", key=" + key + ", msgId=" +
sendResult.getMsgId(), "SUCCESS");
return SendMessageVO.builder()
.msgId(sendResult.getMsgId())
@@ -349,7 +345,7 @@ public class RocketMQAdminClientImpl implements AdminClient
{
.offsetMsgId(sendResult.getOffsetMsgId())
.build();
} catch (Exception e) {
- auditService.record("SEND_MESSAGE", request.getTopic(),
e.getMessage(), "FAILED");
+ recordAudit("SEND_MESSAGE", request.getTopic(), e.getMessage(),
"FAILED");
throw new BusinessException(500, "Failed to send message: " +
e.getMessage());
} finally {
producer.shutdown();
@@ -416,16 +412,16 @@ public class RocketMQAdminClientImpl implements
AdminClient {
groupMapper.updateById(entity);
}
- auditService.record("CREATE_GROUP", groupName,
+ recordAudit("CREATE_GROUP", groupName,
"retryMaxTimes=" + config.getRetryMaxTimes(), "SUCCESS");
group.setId(groupName);
return group;
} catch (BusinessException e) {
- auditService.record("CREATE_GROUP", groupName, e.getMessage(),
"FAILED");
+ recordAudit("CREATE_GROUP", groupName, e.getMessage(), "FAILED");
throw e;
} catch (Exception e) {
- auditService.record("CREATE_GROUP", groupName, e.getMessage(),
"FAILED");
+ recordAudit("CREATE_GROUP", groupName, e.getMessage(), "FAILED");
throw new BusinessException(500, "Failed to create consumer group:
" + e.getMessage());
}
}
@@ -456,12 +452,12 @@ public class RocketMQAdminClientImpl implements
AdminClient {
// Delete from DB
groupMapper.delete(new
LambdaQueryWrapper<RmqGroup>().eq(RmqGroup::getName, name));
- auditService.record("DELETE_GROUP", name, "", "SUCCESS");
+ recordAudit("DELETE_GROUP", name, "", "SUCCESS");
} catch (BusinessException e) {
- auditService.record("DELETE_GROUP", name, e.getMessage(),
"FAILED");
+ recordAudit("DELETE_GROUP", name, e.getMessage(), "FAILED");
throw e;
} catch (Exception e) {
- auditService.record("DELETE_GROUP", name, e.getMessage(),
"FAILED");
+ recordAudit("DELETE_GROUP", name, e.getMessage(), "FAILED");
throw new BusinessException(500, "Failed to delete consumer group:
" + e.getMessage());
}
}
@@ -480,16 +476,25 @@ public class RocketMQAdminClientImpl implements
AdminClient {
return null;
});
}
- auditService.record("RESET_OFFSET", name,
+ recordAudit("RESET_OFFSET", name,
"instanceId=" + instanceId + ", topic=" + topic + ",
timestamp=" + timestamp, "SUCCESS");
} catch (Exception e) {
- auditService.record("RESET_OFFSET", name, e.getMessage(),
"FAILED");
+ recordAudit("RESET_OFFSET", name, e.getMessage(), "FAILED");
throw new BusinessException(500, "Failed to reset offset: " +
e.getMessage());
}
}
// ── Helper methods ──────────────────────────────────────────────────
+ private void recordAudit(String action, String resource, String detail,
String result) {
+ try {
+ auditService.record(action, resource, detail, result);
+ } catch (Exception auditFailure) {
+ log.warn("Failed to record audit action={} resource={}: {}",
action, resource,
+ auditFailure.getMessage());
+ }
+ }
+
/**
* Returns the configured default NameServer address, failing fast when
the studio has no
* RocketMQ endpoint configured (equivalent to the former absent admin
bean).
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQBrokerConfigService.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQBrokerConfigService.java
index 15efe4d2..6dfb4577 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQBrokerConfigService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQBrokerConfigService.java
@@ -60,18 +60,27 @@ public class RocketMQBrokerConfigService {
try {
admin.updateBrokerConfig(brokerAddr, newConfig);
String detail = "brokerAddr=" + brokerAddr + ", config=" +
newConfig;
- auditService.record("UPDATE_BROKER_CONFIG", "CLUSTER:" +
clusterId, detail, "SUCCESS");
+ recordAudit(clusterId, detail, "SUCCESS");
log.info("Broker config updated successfully: {}", brokerAddr);
return null;
} catch (Exception e) {
log.error("Failed to update broker config at {}", brokerAddr,
e);
String detail = "brokerAddr=" + brokerAddr + ", error=" +
e.getMessage();
- auditService.record("UPDATE_BROKER_CONFIG", "CLUSTER:" +
clusterId, detail, "FAILURE");
+ recordAudit(clusterId, detail, "FAILURE");
throw new BusinessException(500, "Failed to update broker
config: " + e.getMessage());
}
});
}
+ private void recordAudit(String clusterId, String detail, String result) {
+ try {
+ auditService.record("UPDATE_BROKER_CONFIG", "CLUSTER:" +
clusterId, detail, result);
+ } catch (Exception auditFailure) {
+ log.warn("Failed to record broker config audit for cluster {}:
{}", clusterId,
+ auditFailure.getMessage());
+ }
+ }
+
private ClusterConfigVO mapToClusterConfigVO(Properties props) {
ClusterConfigVO vo = new ClusterConfigVO();
vo.setFlushDiskType(parseFlushDiskType(props.getProperty("flushDiskType",
"ASYNC_FLUSH")));
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 1e6bcb9c..ca9f193f 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
@@ -194,6 +194,21 @@ class ClusterServiceTest {
verify(clusterRepository).updateConfig(eq("cluster-1"),
any(ClusterConfigVO.class));
}
+ @Test
+ void updateConfigShouldSucceedWhenAuditRecordingFails() {
+
when(clusterRepository.findById("cluster-1")).thenReturn(Optional.of(sampleCluster));
+ doThrow(new IllegalStateException("audit storage
unavailable")).when(auditService)
+ .record(any(), any(), any(), any());
+
+ ClusterConfigUpdateResultVO result =
clusterService.updateClusterConfig(UpdateConfigDTO.builder()
+ .id("cluster-1")
+ .flushDiskType("SYNC_FLUSH")
+ .build());
+
+
assertThat(result.getStatus()).isEqualTo(ClusterConfigUpdateResultVO.Status.SUCCESS);
+ verify(clusterRepository).updateConfig(eq("cluster-1"),
any(ClusterConfigVO.class));
+ }
+
@Test
void updateConfigShouldUpdateMultipleFields() {
when(clusterRepository.findById("cluster-1")).thenReturn(Optional.of(sampleCluster));
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
index 0ad5d522..5877e888 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
@@ -234,6 +234,52 @@ class RocketMQAdminClientImplTest {
verify(adminExt, never()).deleteSubscriptionGroup(anyString(),
anyString(), org.mockito.ArgumentMatchers.anyBoolean());
}
+ @Test
+ void createTopicSucceedsWhenAuditRecordingFails() throws Exception {
+ TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new
MybatisConfiguration(), ""), RmqTopic.class);
+ ClusterInfo clusterInfo = clusterInfoWithMaster();
+ when(adminExt.examineBrokerClusterInfo()).thenReturn(clusterInfo);
+ when(topicMapper.selectOne(any())).thenReturn(null);
+ doNothing().when(adminExt).createAndUpdateTopicConfig(anyString(),
any(TopicConfig.class));
+ doThrow(new RuntimeException("audit db down")).when(auditService)
+ .record(anyString(), anyString(), anyString(), anyString());
+
+ TopicVO topic = new TopicVO();
+ topic.setName("topicA");
+
+ assertThat(adminClient.createTopic(topic).getId()).isEqualTo("topicA");
+ verify(auditService).record("CREATE_TOPIC", "topicA", "queues=8/8",
"SUCCESS");
+ }
+
+ @Test
+ void createTopicPreservesRemoteFailureWhenAuditRecordingFails() throws
Exception {
+ when(adminExt.examineBrokerClusterInfo())
+ .thenThrow(new IllegalStateException("broker unavailable"));
+ doThrow(new RuntimeException("audit db down")).when(auditService)
+ .record(anyString(), anyString(), anyString(), anyString());
+
+ TopicVO topic = new TopicVO();
+ topic.setName("topicA");
+
+ assertThatThrownBy(() -> adminClient.createTopic(topic))
+ .isInstanceOf(BusinessException.class)
+ .hasMessageContaining("Failed to create topic: broker
unavailable");
+ }
+
+ private ClusterInfo clusterInfoWithMaster() {
+ ClusterInfo clusterInfo = new ClusterInfo();
+ Map<String, Set<String>> clusterAddrTable = new HashMap<>();
+ clusterAddrTable.put("cluster-1", new HashSet<>(List.of("broker-1")));
+ clusterInfo.setClusterAddrTable(clusterAddrTable);
+ Map<String, BrokerData> brokerAddrTable = new HashMap<>();
+ BrokerData brokerData = new BrokerData();
+ brokerData.setBrokerName("broker-1");
+ brokerData.setBrokerAddrs(new HashMap<>(Map.of(0L, "10.0.0.1:10911")));
+ brokerAddrTable.put("broker-1", brokerData);
+ clusterInfo.setBrokerAddrTable(brokerAddrTable);
+ return clusterInfo;
+ }
+
@Test
void sendMessageShouldNotFailWhenAuditRecordingFails() throws Exception {
when(properties.getNamesrvAddr()).thenReturn("10.0.0.1:9876");
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQBrokerConfigServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQBrokerConfigServiceTest.java
new file mode 100644
index 00000000..83b8fe4d
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQBrokerConfigServiceTest.java
@@ -0,0 +1,73 @@
+/*
+ * 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.provider.apache;
+
+import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
+import org.apache.rocketmq.studio.common.exception.BusinessException;
+import org.apache.rocketmq.studio.ops.audit.AuditService;
+import org.apache.rocketmq.tools.admin.DefaultMQAdminExt;
+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.Properties;
+
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.lenient;
+
+@ExtendWith(MockitoExtension.class)
+class RocketMQBrokerConfigServiceTest {
+
+ @Mock
+ private MqAdminExtFactory adminFactory;
+ @Mock
+ private RocketMQProperties properties;
+ @Mock
+ private DefaultMQAdminExt adminExt;
+ @Mock
+ private AuditService auditService;
+
+ private RocketMQBrokerConfigService brokerConfigService;
+
+ @BeforeEach
+ void setUp() {
+
lenient().when(properties.getNamesrvAddr()).thenReturn("10.0.0.1:9876");
+ lenient().when(adminFactory.execute(anyString(), any(),
any())).thenAnswer(invocation ->
+
invocation.<MqAdminExtFactory.AdminAction<Object>>getArgument(2).apply(adminExt));
+ brokerConfigService = new RocketMQBrokerConfigService(adminFactory,
properties, auditService);
+ }
+
+ @Test
+ void updateSucceedsWhenAuditRecordingFails() throws Exception {
+ Properties config = new Properties();
+ config.setProperty("flushDiskType", "ASYNC_FLUSH");
+ doNothing().when(adminExt).updateBrokerConfig("broker-a:10911",
config);
+ doThrow(new IllegalStateException("audit db down")).when(auditService)
+ .record(anyString(), anyString(), anyString(), anyString());
+
+ brokerConfigService.updateBrokerConfig("broker-a:10911", "cluster-a",
config);
+ }
+
+ @Test
+ void updatePreservesBrokerFailureWhenAuditRecordingFails() throws
Exception {
+ Properties config = new Properties();
+ doThrow(new IllegalStateException("broker unavailable")).when(adminExt)
+ .updateBrokerConfig("broker-a:10911", config);
+ doThrow(new IllegalStateException("audit db down")).when(auditService)
+ .record(anyString(), anyString(), anyString(), anyString());
+
+ assertThatThrownBy(() ->
brokerConfigService.updateBrokerConfig("broker-a:10911", "cluster-a", config))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Failed to update broker config: broker
unavailable");
+ }
+}
