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 3c1382a28 fix(alerts): isolate bulk rule item transactions (#7221)
3c1382a28 is described below
commit 3c1382a28823a3a3dcfbef152df5db68c9ff1c7a
Author: aias00 <[email protected]>
AuthorDate: Mon Oct 26 16:34:45 2026 +0800
fix(alerts): isolate bulk rule item transactions (#7221)
* fix(alerts): isolate bulk rule item transactions
---
.../rocketmq/studio/ops/alert/AlertService.java | 71 +++++++++++-----
.../alert/AlertBulkMutationTransactionTest.java | 97 ++++++++++++++++++++++
.../ops/alert/AlertRuleStateLifecycleTest.java | 3 +-
.../ops/alert/AlertRuleTransferServiceTest.java | 3 +-
.../ops/alert/AlertServiceDefaultRulesTest.java | 6 +-
.../studio/ops/alert/AlertServiceTest.java | 6 +-
6 files changed, 158 insertions(+), 28 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertService.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertService.java
index 0c26bfec7..7c958018e 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertService.java
@@ -26,6 +26,9 @@ import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.util.StringUtils;
@@ -62,6 +65,7 @@ public class AlertService {
private final AlertRuleAssetService alertRuleAssetService;
private final OperationAuditService operationAuditService;
private final MetricProfileService metricProfileService;
+ private final PlatformTransactionManager transactionManager;
public List<AlertRuleVO> listRules() {
@@ -334,7 +338,6 @@ public class AlertService {
recordAudit("DELETE_ALERT_RULE", "ALERT_RULE", String.valueOf(id),
null, null);
}
- @Transactional
public AlertRuleBulkResultVO bulkToggleRules(List<Long> ids, boolean
enabled) {
List<Long> normalizedIds = normalizeBulkIds(ids);
Map<Long, AlertRuleVO> rulesById = rulesById(normalizedIds);
@@ -349,16 +352,20 @@ public class AlertService {
}
try {
rule.setEnabled(enabled);
- // replaceRule performs an existence-checked update; a rule
deleted between the
- // snapshot load and this call is reported as a failure
instead of being recreated.
- if (!alertRepository.replaceRule(rule)) {
+ if (!executeBulkMutation(() -> {
+ // A concurrently deleted rule must not be recreated.
+ if (!alertRepository.replaceRule(rule)) {
+ return false;
+ }
+ if (!enabled) {
+ alertStateRepository.deleteByRuleId(id);
+ }
+ auditRule("TOGGLE_ALERT_RULE", rule, "enabled=" + enabled
+ ", bulk=true");
+ return true;
+ })) {
failures.put(id, "Alert rule not found");
continue;
}
- if (!enabled) {
- alertStateRepository.deleteByRuleId(id);
- }
- auditRule("TOGGLE_ALERT_RULE", rule, "enabled=" + enabled + ",
bulk=true");
succeeded.add(id);
updated.add(rule);
} catch (RuntimeException failure) {
@@ -369,26 +376,29 @@ public class AlertService {
.succeededIds(succeeded).failures(failures).updatedRules(updated).build();
}
- @Transactional
public AlertRuleBulkResultVO bulkToggleRules(AlertDomain domain,
List<Long> ids, boolean enabled) {
requireDomain(domain);
return bulkUpdateRules(domain, ids, rule -> rule.setEnabled(enabled),
"TOGGLE_ALERT_RULE", "enabled=" + enabled + ", bulk=true");
}
- @Transactional
public AlertRuleBulkResultVO bulkDeleteRules(List<Long> ids) {
List<Long> normalizedIds = normalizeBulkIds(ids);
List<Long> succeeded = new ArrayList<>();
Map<Long, String> failures = new LinkedHashMap<>();
for (Long id : normalizedIds) {
try {
- if (!alertRepository.deleteRule(id)) {
+ if (!executeBulkMutation(() -> {
+ if (!alertRepository.deleteRule(id)) {
+ return false;
+ }
+ alertStateRepository.deleteByRuleId(id);
+ recordAudit("DELETE_ALERT_RULE", "ALERT_RULE",
String.valueOf(id), null, "bulk=true");
+ return true;
+ })) {
failures.put(id, "Alert rule not found");
continue;
}
- alertStateRepository.deleteByRuleId(id);
- recordAudit("DELETE_ALERT_RULE", "ALERT_RULE",
String.valueOf(id), null, "bulk=true");
succeeded.add(id);
} catch (RuntimeException failure) {
failures.put(id, failure.getMessage() == null ? "Delete
failed" : failure.getMessage());
@@ -398,7 +408,6 @@ public class AlertService {
.succeededIds(succeeded).failures(failures).updatedRules(List.of()).build();
}
- @Transactional
public AlertRuleBulkResultVO bulkDeleteRules(AlertDomain domain,
List<Long> ids) {
requireDomain(domain);
List<Long> normalizedIds = normalizeBulkIds(ids);
@@ -412,12 +421,17 @@ public class AlertService {
continue;
}
try {
- if (!alertRepository.deleteRule(id)) {
+ if (!executeBulkMutation(() -> {
+ if (!alertRepository.deleteRule(id)) {
+ return false;
+ }
+ alertStateRepository.deleteByRuleId(id);
+ recordAudit("DELETE_ALERT_RULE", "ALERT_RULE",
String.valueOf(id), null, "bulk=true");
+ return true;
+ })) {
failures.put(id, "Alert rule not found");
continue;
}
- alertStateRepository.deleteByRuleId(id);
- recordAudit("DELETE_ALERT_RULE", "ALERT_RULE",
String.valueOf(id), null, "bulk=true");
succeeded.add(id);
} catch (RuntimeException failure) {
failures.put(id, failure.getMessage() == null ? "Delete
failed" : failure.getMessage());
@@ -442,14 +456,19 @@ public class AlertService {
}
try {
update.accept(rule);
- if (!alertRepository.replaceRule(rule)) {
+ if (!executeBulkMutation(() -> {
+ if (!alertRepository.replaceRule(rule)) {
+ return false;
+ }
+ if (!rule.isEnabled()) {
+ alertStateRepository.deleteByRuleId(id);
+ }
+ auditRule(auditOperation, rule, auditDetail);
+ return true;
+ })) {
failures.put(id, "Alert rule not found");
continue;
}
- if (!rule.isEnabled()) {
- alertStateRepository.deleteByRuleId(id);
- }
- auditRule(auditOperation, rule, auditDetail);
succeeded.add(id);
updated.add(rule);
} catch (RuntimeException failure) {
@@ -460,6 +479,14 @@ public class AlertService {
.succeededIds(succeeded).failures(failures).updatedRules(updated).build();
}
+ private boolean executeBulkMutation(java.util.function.BooleanSupplier
mutation) {
+ TransactionTemplate transaction = new
TransactionTemplate(transactionManager);
+ // Catch failures only after rollback; a failed item must not affect
other items,
+ // even when the bulk operation is invoked from an existing
transaction.
+
transaction.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
+ return Boolean.TRUE.equals(transaction.execute(status ->
mutation.getAsBoolean()));
+ }
+
private AlertRuleVO findRuleInDomain(AlertDomain domain, Long id) {
validateRuleId(id);
return alertRepository.findRuleById(id)
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertBulkMutationTransactionTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertBulkMutationTransactionTest.java
new file mode 100644
index 000000000..308d0fad9
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertBulkMutationTransactionTest.java
@@ -0,0 +1,97 @@
+/*
+ * 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.ops.alert;
+
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.test.context.ActiveProfiles;
+import org.springframework.test.context.bean.override.mockito.MockitoSpyBean;
+
+import java.util.List;
+import java.util.UUID;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.doAnswer;
+
+@SpringBootTest
+@ActiveProfiles("dev")
+class AlertBulkMutationTransactionTest {
+ @Autowired
+ private AlertService service;
+ @Autowired
+ private AlertRepository rules;
+ @Autowired
+ private JdbcTemplate jdbc;
+ @MockitoSpyBean
+ private AlertStateRepository states;
+
+ @ParameterizedTest
+ @CsvSource({"false,false", "false,true", "true,false", "true,true"})
+ void failedItemRollsBackWhileOtherItemsCommitTest(boolean scoped, boolean
delete) {
+ Long failedId = insertRuleTest();
+ Long healthyId = insertRuleTest();
+ try {
+ doAnswer(invocation -> {
+ invocation.callRealMethod();
+ throw new IllegalStateException("state cleanup failed");
+ }).when(states).deleteByRuleId(failedId);
+ List<Long> ids = List.of(failedId, healthyId);
+ AlertRuleBulkResultVO result;
+ if (delete) {
+ result = scoped ?
service.bulkDeleteRules(AlertDomain.BUSINESS, ids)
+ : service.bulkDeleteRules(ids);
+ } else {
+ result = scoped ?
service.bulkToggleRules(AlertDomain.BUSINESS, ids, false)
+ : service.bulkToggleRules(ids, false);
+ }
+ assertThat(result.getSucceededIds()).containsExactly(healthyId);
+ assertThat(result.getFailures()).containsOnlyKeys(failedId);
+ assertThat(result.getUpdatedRules()).extracting(AlertRuleVO::getId)
+ .containsExactlyElementsOf(delete ? List.of() :
List.of(healthyId));
+ assertThat(rules.findRuleById(failedId)).isPresent();
+
assertThat(rules.findRuleById(failedId).orElseThrow().isEnabled()).isTrue();
+ assertThat(stateCountTest(failedId)).isEqualTo(1);
+ if (delete) {
+ assertThat(rules.findRuleById(healthyId)).isEmpty();
+ } else {
+
assertThat(rules.findRuleById(healthyId).orElseThrow().isEnabled()).isFalse();
+ }
+ assertThat(stateCountTest(healthyId)).isZero();
+ } finally {
+ for (Long id : List.of(failedId, healthyId)) {
+ jdbc.update("DELETE FROM rmq_alert_state WHERE rule_id = ?",
id);
+ rules.deleteRule(id);
+ }
+ }
+ }
+
+ private Long insertRuleTest() {
+ AlertRuleVO rule =
rules.insertRule(AlertRuleVO.builder().name("bulk-tx-" + UUID.randomUUID())
+
.domain(AlertDomain.BUSINESS).metric("consumer.lag.total").operator(">")
+
.threshold(10).enabled(true).instanceId("local").consumerGroup("orders-" +
UUID.randomUUID()).build());
+ jdbc.update("INSERT INTO rmq_alert_state (rule_id, fingerprint,
status, consecutive_hits) "
+ + "VALUES (?, ?, 'FIRING', 1)", rule.getId(), "bulk-tx-state");
+ return rule.getId();
+ }
+
+ private int stateCountTest(Long id) {
+ return jdbc.queryForObject("SELECT COUNT(*) FROM rmq_alert_state WHERE
rule_id = ?", Integer.class, id);
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertRuleStateLifecycleTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertRuleStateLifecycleTest.java
index 1c2bbeb03..37c00cb59 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertRuleStateLifecycleTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertRuleStateLifecycleTest.java
@@ -45,7 +45,8 @@ class AlertRuleStateLifecycleTest {
@BeforeEach
void setUpTest() {
service = new AlertService(rules, states, new AlertRuleAssetService(),
- mock(OperationAuditService.class), new
MetricProfileService(new PrometheusProperties()));
+ mock(OperationAuditService.class), new
MetricProfileService(new PrometheusProperties()),
+
org.mockito.Mockito.mock(org.springframework.transaction.PlatformTransactionManager.class));
storedRule.set(rule(true, 100, "Lag"));
runtime.set(AlertRuleRuntimeVO.builder().ruleId(4L).fingerprint("active-episode")
.status(AlertStateStatus.FIRING).lastNotifiedAt(LocalDateTime.of(2026, 9, 1,
12, 0))
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertRuleTransferServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertRuleTransferServiceTest.java
index b9849bb84..8d790ce24 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertRuleTransferServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertRuleTransferServiceTest.java
@@ -107,7 +107,8 @@ class AlertRuleTransferServiceTest {
AlertRepository repository = mock(AlertRepository.class);
OperationAuditService auditService = mock(OperationAuditService.class);
AlertService realService = new AlertService(repository,
mock(AlertStateRepository.class),
- new AlertRuleAssetService(), auditService, new
MetricProfileService(new PrometheusProperties()));
+ new AlertRuleAssetService(), auditService, new
MetricProfileService(new PrometheusProperties()),
+
org.mockito.Mockito.mock(org.springframework.transaction.PlatformTransactionManager.class));
AlertRuleTransferService realTransfer = new
AlertRuleTransferService(realService, metricCatalogService);
lenient().when(repository.insertRule(any(AlertRuleVO.class)))
.thenAnswer(invocation -> invocation.getArgument(0));
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertServiceDefaultRulesTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertServiceDefaultRulesTest.java
index a31d3bce5..0d23e3e23 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertServiceDefaultRulesTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertServiceDefaultRulesTest.java
@@ -42,7 +42,8 @@ class AlertServiceDefaultRulesTest {
AlertService service = new AlertService(repository,
Mockito.mock(AlertStateRepository.class),
new AlertRuleAssetService(),
Mockito.mock(OperationAuditService.class),
- new MetricProfileService(new PrometheusProperties()));
+ new MetricProfileService(new PrometheusProperties()),
+
org.mockito.Mockito.mock(org.springframework.transaction.PlatformTransactionManager.class));
String yaml = service.exportPrometheusRulesYaml();
int ruleCount = countRules(yaml);
@@ -56,7 +57,8 @@ class AlertServiceDefaultRulesTest {
AlertService service = new AlertService(repository,
Mockito.mock(AlertStateRepository.class),
new AlertRuleAssetService(),
Mockito.mock(OperationAuditService.class),
- new MetricProfileService(new PrometheusProperties()));
+ new MetricProfileService(new PrometheusProperties()),
+
org.mockito.Mockito.mock(org.springframework.transaction.PlatformTransactionManager.class));
String yaml = service.exportPrometheusRulesYaml();
assertTrue(yaml.contains("rocketmq-broker.rules"));
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertServiceTest.java
index cd726daa2..793ca52de 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertServiceTest.java
@@ -70,7 +70,8 @@ class AlertServiceTest {
@org.junit.jupiter.api.BeforeEach
void setUpTest() {
alertService = new AlertService(alertRepository, alertStateRepository,
new AlertRuleAssetService(),
- operationAuditService, new MetricProfileService(new
PrometheusProperties()));
+ operationAuditService, new MetricProfileService(new
PrometheusProperties()),
+
org.mockito.Mockito.mock(org.springframework.transaction.PlatformTransactionManager.class));
}
@Test
@@ -566,7 +567,8 @@ class AlertServiceTest {
PrometheusProperties properties = new PrometheusProperties();
properties.setProfile(profileId);
return new AlertService(alertRepository, alertStateRepository, new
AlertRuleAssetService(),
- operationAuditService, new MetricProfileService(properties));
+ operationAuditService, new MetricProfileService(properties),
+
org.mockito.Mockito.mock(org.springframework.transaction.PlatformTransactionManager.class));
}
private MetricProfileVO.MetricMappingVO activeLagMapping(String profileId)
{
