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 94d2f86f7 [ISSUE #2914][ISSUE #2954][ISSUE #2953] fix(server):
consolidate persistence and audit correctness (#2919)
94d2f86f7 is described below
commit 94d2f86f7cabe0fa95ad0b1c04d3510a773c3915
Author: shown <[email protected]>
AuthorDate: Fri Sep 4 14:19:59 2026 +0800
[ISSUE #2914][ISSUE #2954][ISSUE #2953] fix(server): consolidate
persistence and audit correctness (#2919)
* [ISSUE #2914] Prevent stale resource recreation
* [ISSUE #2954] Preserve nested audit detail values
* [ISSUE #2953] Isolate audit persistence failures
---
.../studio/audit/OperationAuditService.java | 11 ++++-
.../cluster/k8s/MybatisPlusK8sCertRepository.java | 2 +-
.../MybatisPlusCloudCredentialRepository.java | 2 +-
.../studio/audit/OperationAuditServiceTest.java | 13 +++++
.../k8s/MybatisPlusK8sCertRepositoryTest.java | 16 ++++++-
.../MybatisPlusCloudCredentialRepositoryTest.java | 15 +++++-
.../pages/ops/__tests__/auditPresentation.test.ts | 15 ++++++
web/src/pages/ops/auditPresentation.ts | 55 +++++++++++++++++++++-
8 files changed, 122 insertions(+), 7 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/audit/OperationAuditService.java
b/server/src/main/java/org/apache/rocketmq/studio/audit/OperationAuditService.java
index c52baff5c..9959cad75 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/audit/OperationAuditService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/audit/OperationAuditService.java
@@ -48,7 +48,14 @@ public class OperationAuditService {
LocalDateTime now = LocalDateTime.now();
audit.setGmtCreate(now);
audit.setGmtModified(now);
- auditMapper.insert(audit);
- log.debug("Audit recorded: {} {} {}", operation, resourceType,
resourceName);
+ try {
+ auditMapper.insert(audit);
+ log.debug("Audit recorded: {} {} {}", operation, resourceType,
resourceName);
+ } catch (RuntimeException auditFailure) {
+ // Audit is observational. A failed sink must not turn an
operation that already
+ // completed at a broker or provider into an API failure that
callers may retry.
+ log.warn("Failed to record audit operation={} resourceType={}
resource={}: {}",
+ operation, resourceType, resourceName,
auditFailure.getMessage());
+ }
}
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/k8s/MybatisPlusK8sCertRepository.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/k8s/MybatisPlusK8sCertRepository.java
index 7ab9b5769..f1839ba9c 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/k8s/MybatisPlusK8sCertRepository.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/k8s/MybatisPlusK8sCertRepository.java
@@ -66,7 +66,7 @@ public class MybatisPlusK8sCertRepository implements
K8sCertRepository {
@Transactional
public K8sCertVO save(K8sCertVO cert) {
RmqK8sCertificate entity = toEntity(cert);
- if (entity.getId() != null && certMapper.selectById(entity.getId()) !=
null) {
+ if (entity.getId() != null) {
if (certMapper.updateById(entity) == 0) {
throw new BusinessException(409,
"Certificate update was not applied: " +
entity.getId());
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/credential/MybatisPlusCloudCredentialRepository.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/credential/MybatisPlusCloudCredentialRepository.java
index f9676ffc8..b010fb3e0 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/credential/MybatisPlusCloudCredentialRepository.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/credential/MybatisPlusCloudCredentialRepository.java
@@ -85,7 +85,7 @@ public class MybatisPlusCloudCredentialRepository implements
CloudCredentialRepo
@Override
public CloudCredentialVO save(CloudCredentialVO credential) {
RmqCloudCredential entity = toEntity(credential);
- if (entity.getId() != null &&
credentialMapper.selectById(entity.getId()) != null) {
+ if (entity.getId() != null) {
if (credentialMapper.updateById(entity) == 0) {
throw new BusinessException(409,
"Cloud credential update was not applied: " +
entity.getId());
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/audit/OperationAuditServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/audit/OperationAuditServiceTest.java
index 645365a8a..7cab90d10 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/audit/OperationAuditServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/audit/OperationAuditServiceTest.java
@@ -28,6 +28,8 @@ import org.mockito.Mock;
import org.mockito.junit.jupiter.MockitoExtension;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.verify;
@ExtendWith(MockitoExtension.class)
@@ -66,4 +68,15 @@ class OperationAuditServiceTest {
assertThat(captor.getValue().getOperator())
.isEqualTo(AuthenticatedUserContext.SYSTEM_ACTOR);
}
+
+ @Test
+ void recordShouldNotPropagateAuditPersistenceFailures() {
+ OperationAuditService service = new OperationAuditService(auditMapper);
+ doThrow(new IllegalStateException("audit database unavailable"))
+
.when(auditMapper).insert(org.mockito.ArgumentMatchers.any(RmqOperationAudit.class));
+
+ assertThatCode(() -> service.record("DIRECT_CONSUME_MESSAGE",
"MESSAGE", "msg-1",
+ "cluster-a", "result=SUCCESS", "SUCCESS", null))
+ .doesNotThrowAnyException();
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/k8s/MybatisPlusK8sCertRepositoryTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/k8s/MybatisPlusK8sCertRepositoryTest.java
index 72d6f406d..00d3b3d29 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/k8s/MybatisPlusK8sCertRepositoryTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/k8s/MybatisPlusK8sCertRepositoryTest.java
@@ -26,6 +26,8 @@ import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.Mockito.mock;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
class MybatisPlusK8sCertRepositoryTest {
@@ -50,7 +52,6 @@ class MybatisPlusK8sCertRepositoryTest {
@Test
void saveShouldReportALostConcurrentUpdate() {
RmqK8sCertificateMapper mapper = mock(RmqK8sCertificateMapper.class);
- when(mapper.selectById(1L)).thenReturn(certificate());
when(mapper.updateById(any(RmqK8sCertificate.class))).thenReturn(0);
K8sCertVO cert = K8sCertVO.builder().k8sId("broker").build();
cert.setId(1L);
@@ -62,6 +63,19 @@ class MybatisPlusK8sCertRepositoryTest {
((BusinessException) error).getCode()).isEqualTo(409));
}
+ @Test
+ void saveShouldNotReinsertACertificateDeletedConcurrently() {
+ RmqK8sCertificateMapper mapper = mock(RmqK8sCertificateMapper.class);
+ when(mapper.updateById(any(RmqK8sCertificate.class))).thenReturn(0);
+ K8sCertVO cert = K8sCertVO.builder().k8sId("broker").build();
+ cert.setId(1L);
+
+ assertThatThrownBy(() -> repository(mapper).save(cert))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Certificate update was not applied: 1");
+ verify(mapper, never()).insert(any(RmqK8sCertificate.class));
+ }
+
@Test
void deleteByIdShouldReportWhetherARowWasRemoved() {
RmqK8sCertificateMapper mapper = mock(RmqK8sCertificateMapper.class);
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/credential/MybatisPlusCloudCredentialRepositoryTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/credential/MybatisPlusCloudCredentialRepositoryTest.java
index 7ee62d058..b9b9ffa00 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/credential/MybatisPlusCloudCredentialRepositoryTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/credential/MybatisPlusCloudCredentialRepositoryTest.java
@@ -38,6 +38,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.Mockito.verify;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.when;
@ExtendWith(MockitoExtension.class)
@@ -54,7 +55,6 @@ class MybatisPlusCloudCredentialRepositoryTest {
CloudCredentialVO credential = new CloudCredentialVO();
credential.setId(1L);
credential.setVendor(InstanceVendor.ALIYUN);
- when(credentialMapper.selectById(1L)).thenReturn(entity(1L, "cred-1",
"ALIYUN"));
when(credentialMapper.updateById(any(RmqCloudCredential.class))).thenReturn(0);
assertThatThrownBy(() -> repository.save(credential))
@@ -63,6 +63,19 @@ class MybatisPlusCloudCredentialRepositoryTest {
.satisfies(error -> assertThat(((BusinessException)
error).getCode()).isEqualTo(409));
}
+ @Test
+ void saveShouldNotReinsertACredentialDeletedConcurrently() {
+ CloudCredentialVO credential = new CloudCredentialVO();
+ credential.setId(1L);
+ credential.setVendor(InstanceVendor.ALIYUN);
+
when(credentialMapper.updateById(any(RmqCloudCredential.class))).thenReturn(0);
+
+ assertThatThrownBy(() -> repository.save(credential))
+ .isInstanceOf(BusinessException.class)
+ .hasMessage("Cloud credential update was not applied: 1");
+ verify(credentialMapper,
never()).insert(any(RmqCloudCredential.class));
+ }
+
@Test
void findByIdShouldMapValidPersistedVendor() {
when(credentialMapper.selectById(2L)).thenReturn(entity(2L,
"cred-valid", "ALIYUN"));
diff --git a/web/src/pages/ops/__tests__/auditPresentation.test.ts
b/web/src/pages/ops/__tests__/auditPresentation.test.ts
index b030e9a3d..cfb7af35d 100644
--- a/web/src/pages/ops/__tests__/auditPresentation.test.ts
+++ b/web/src/pages/ops/__tests__/auditPresentation.test.ts
@@ -116,6 +116,21 @@ describe('audit presentation helpers', () => {
]);
});
+ it('preserves commas inside nested configuration and quoted values', () => {
+ expect(
+ parseAuditDetail(
+ 'brokerAddr=10.0.0.1:10911, config={flushDiskType=SYNC_FLUSH,
fileReservedTime=72}, note="primary, synchronous"',
+ ),
+ ).toEqual([
+ { label: 'brokerAddr', value: '10.0.0.1:10911' },
+ {
+ label: 'config',
+ value: '{flushDiskType=SYNC_FLUSH, fileReservedTime=72}',
+ },
+ { label: 'note', value: '"primary, synchronous"' },
+ ]);
+ });
+
it('keeps free-form details intact when they are not key-value lists', () =>
{
expect(parseAuditDetail('Removed stale Proxy address
10.0.30.9:8081')).toEqual([
{ label: '', value: 'Removed stale Proxy address 10.0.30.9:8081' },
diff --git a/web/src/pages/ops/auditPresentation.ts
b/web/src/pages/ops/auditPresentation.ts
index 1980ce7c6..c19fd274d 100644
--- a/web/src/pages/ops/auditPresentation.ts
+++ b/web/src/pages/ops/auditPresentation.ts
@@ -272,7 +272,8 @@ export function parseAuditDetail(detail: string | null |
undefined): AuditDetail
const text = detail?.trim();
if (!text) return [];
- const parts = text.split(/\s*,\s*/).filter(Boolean);
+ const parts = splitTopLevelDetailFields(text);
+ if (!parts) return [{ label: '', value: text }];
if (parts.length <= 1) return [{ label: '', value: text }];
const tokens = parts.map((part) => {
@@ -290,3 +291,55 @@ export function parseAuditDetail(detail: string | null |
undefined): AuditDetail
}
return tokens as AuditDetailToken[];
}
+
+const DETAIL_DELIMITERS: Record<string, string> = {
+ '{': '}',
+ '[': ']',
+ '(': ')',
+};
+
+function splitTopLevelDetailFields(text: string): string[] | null {
+ const fields: string[] = [];
+ const expectedClosings: string[] = [];
+ let fieldStart = 0;
+ let quote = '';
+ let escaped = false;
+
+ for (let index = 0; index < text.length; index += 1) {
+ const char = text[index];
+ if (quote) {
+ if (escaped) {
+ escaped = false;
+ } else if (char === '\\') {
+ escaped = true;
+ } else if (char === quote) {
+ quote = '';
+ }
+ continue;
+ }
+
+ if (char === '"' || char === "'") {
+ quote = char;
+ continue;
+ }
+ const closing = DETAIL_DELIMITERS[char];
+ if (closing) {
+ expectedClosings.push(closing);
+ continue;
+ }
+ if (char === '}' || char === ']' || char === ')') {
+ if (expectedClosings.pop() !== char) return null;
+ continue;
+ }
+ if (char === ',' && expectedClosings.length === 0) {
+ const field = text.slice(fieldStart, index).trim();
+ if (field) fields.push(field);
+ fieldStart = index + 1;
+ }
+ }
+
+ if (quote || expectedClosings.length > 0) return null;
+ const field = text.slice(fieldStart).trim();
+ if (field) fields.push(field);
+ return fields;
+}
