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 1b59bbd4b fix(alert): isolate each reconciliation state save in its
own transaction (#3039)
1b59bbd4b is described below
commit 1b59bbd4b6be94e42178a7a44fabaf2038b8bfc6
Author: cyberslack_lee <[email protected]>
AuthorDate: Tue Sep 15 19:30:45 2026 +0800
fix(alert): isolate each reconciliation state save in its own transaction
(#3039)
* Signed-off-by: enkilee <[email protected]>
fix Reconciliation 循环中单次 emitLifecycleEvent 失败回滚所有已保存的状态
* Signed-off-by: enkilee <[email protected]>
fix using comment suggestion
* Fix: fix code by comment
---
.../studio/ops/alert/NativeAlertProcessor.java | 37 +++++-
.../studio/ops/alert/NativeAlertProcessorTest.java | 146 +++++++++++++++++++--
2 files changed, 166 insertions(+), 17 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NativeAlertProcessor.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NativeAlertProcessor.java
index 2ae7b22d1..e4dacd3a8 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NativeAlertProcessor.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NativeAlertProcessor.java
@@ -23,7 +23,10 @@ import
org.apache.rocketmq.studio.cluster.metrics.MetricCollectionScope;
import org.apache.rocketmq.studio.cluster.metrics.MetricSample;
import org.apache.rocketmq.studio.common.domain.enums.AlertLevel;
import org.springframework.stereotype.Component;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
+import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.util.StringUtils;
import java.time.Instant;
@@ -50,6 +53,7 @@ public class NativeAlertProcessor {
private final AlertRepository alertRepository;
private final NotificationOutboxService notificationOutboxService;
private final AlertNotificationSuppressionService
notificationSuppressionService;
+ private final PlatformTransactionManager transactionManager;
public void process(List<MetricSample> samples) {
processSamples(samples);
@@ -112,10 +116,14 @@ public class NativeAlertProcessor {
.map(rule -> new AlertStateKey(rule.getId(),
AlertFingerprint.of(rule.getId(),
sample.instanceId(), sample.labels()))))
.collect(Collectors.toSet());
- Map<Long, AlertRuleVO> byId =
rules.stream().collect(Collectors.toMap(AlertRuleVO::getId, rule -> rule));
+ Map<Long, AlertRuleVO> byId =
rules.stream().collect(Collectors.toMap(AlertRuleVO::getId, rule -> rule,
+ (left, right) -> left));
Instant resolvedAt =
samples.stream().filter(scope::contains).map(MetricSample::collectedAt).max(Instant::compareTo)
.orElseGet(Instant::now);
AlertEvaluationResult clear = new AlertEvaluationResult(true, false,
null, MetricAvailability.AVAILABLE);
+ TransactionTemplate isolatedTx = new
TransactionTemplate(transactionManager);
+ isolatedTx.setPropagationBehavior(Propagation.REQUIRES_NEW.value());
+ int failedLifecycleEmits = 0;
for (ActiveAlertState active : stateRepository.findActive(scope,
rules)) {
if (presentKeys.contains(active.key())) {
continue;
@@ -130,11 +138,30 @@ public class NativeAlertProcessor {
if (update.transition() != AlertStateTransition.RESOLVED) {
continue;
}
- if (!stateRepository.save(active.key(), update.state())) {
- continue;
+ try {
+ // save + emit run inside a single REQUIRES_NEW
sub-transaction so that a
+ // transient emit failure rolls back both the state change and
the event,
+ // preventing orphaned states or events. The next collection
cycle will
+ // re-evaluate this active state and re-attempt the emit.
+ // save() may still return false inside the sub-transaction
when a
+ // concurrent writer (e.g. a user ACK) won the optimistic
race, so the
+ // emit is gated on the save result: an alert that already
moved on must
+ // not receive a RESOLVED event.
+ isolatedTx.executeWithoutResult(txStatus -> {
+ if (!stateRepository.save(active.key(), update.state())) {
+ return;
+ }
+ emitLifecycleEvent(rule, active.key(), update,
scope.domain(),
+ active.instanceId(), rule.getMetric(),
active.labels(), resolvedAt);
+ });
+ } catch (RuntimeException error) {
+ failedLifecycleEmits++;
+ log.warn("Native alert reconcile lifecycle emit failed:
ruleId={}, fingerprint={}, cause={}",
+ rule.getId(), active.key().fingerprint(),
error.getClass().getSimpleName());
}
- emitLifecycleEvent(rule, active.key(), update, scope.domain(),
active.instanceId(), rule.getMetric(),
- active.labels(), resolvedAt);
+ }
+ if (failedLifecycleEmits > 0) {
+ log.warn("Native alert reconcile completed with {} failed
lifecycle emit(s)", failedLifecycleEmits);
}
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertProcessorTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertProcessorTest.java
index 54c462407..552a78a34 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertProcessorTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertProcessorTest.java
@@ -21,6 +21,9 @@ import
org.apache.rocketmq.studio.cluster.metrics.MetricCollectionScope;
import org.apache.rocketmq.studio.cluster.metrics.MetricSample;
import org.apache.rocketmq.studio.cluster.metrics.MetricSnapshotRepository;
import org.junit.jupiter.api.Test;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.TransactionDefinition;
+import org.springframework.transaction.TransactionStatus;
import java.time.Instant;
import java.util.HashMap;
@@ -33,6 +36,7 @@ import static org.assertj.core.api.Assertions.assertThat;
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.anyMap;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
@@ -103,7 +107,7 @@ class NativeAlertProcessorTest {
NativeAlertProcessor processor = new NativeAlertProcessor(service,
evaluationService,
new AlertStateMachine(), mock(AlertStateRepository.class),
mock(AlertRepository.class),
- mock(NotificationOutboxService.class), suppression());
+ mock(NotificationOutboxService.class), suppression(),
mockTxManager());
assertThatThrownBy(() -> processor.process(List.of(sample("orders"))))
.isInstanceOf(AssertionError.class)
@@ -354,7 +358,7 @@ class NativeAlertProcessorTest {
NativeAlertEvaluationService evaluationService = new
NativeAlertEvaluationService(new AlertRuleEvaluator(),
new AlertStateMachine(), states, snapshots, alerts, outbox,
suppression);
return new NativeAlertProcessor(service, evaluationService, new
AlertStateMachine(), states, alerts, outbox,
- suppression);
+ suppression, mockTxManager());
}
private static NativeAlertProcessor processor(AlertService service,
AlertStateRepository states,
@@ -431,11 +435,11 @@ class NativeAlertProcessorTest {
when(alerts.saveAlert(any(SystemAlertVO.class))).thenAnswer(invocation
-> invocation.getArgument(0));
NotificationOutboxService outbox =
mock(NotificationOutboxService.class);
- new NativeAlertProcessor(service,
+ NativeAlertProcessor processor = new NativeAlertProcessor(service,
new NativeAlertEvaluationService(new AlertRuleEvaluator(), new
AlertStateMachine(), states,
mock(MetricSnapshotRepository.class), alerts, outbox,
suppression()),
- new AlertStateMachine(), states, alerts, outbox, suppression())
- .processSuccessfulCollection(new
MetricCollectionScope(AlertDomain.BUSINESS, "local",
+ new AlertStateMachine(), states, alerts, outbox,
suppression(), mockTxManager());
+ processor.processSuccessfulCollection(new
MetricCollectionScope(AlertDomain.BUSINESS, "local",
java.util.Set.of("consumer.lag.total")), List.of());
org.mockito.ArgumentCaptor<AlertRuleState> state =
org.mockito.ArgumentCaptor.forClass(AlertRuleState.class);
@@ -466,12 +470,13 @@ class NativeAlertProcessorTest {
when(states.findActive(any(MetricCollectionScope.class),
eq(List.of(rule)))).thenReturn(List.of(active));
AlertRepository alerts = mock(AlertRepository.class);
- new NativeAlertProcessor(service,
+ NativeAlertProcessor processor = new NativeAlertProcessor(service,
new NativeAlertEvaluationService(new AlertRuleEvaluator(), new
AlertStateMachine(), states,
mock(MetricSnapshotRepository.class), alerts,
mock(NotificationOutboxService.class),
suppression()),
- new AlertStateMachine(), states, alerts,
mock(NotificationOutboxService.class), suppression())
- .processSuccessfulCollection(new
MetricCollectionScope(AlertDomain.BUSINESS, "local",
+ new AlertStateMachine(), states, alerts,
mock(NotificationOutboxService.class), suppression(),
+ mockTxManager());
+ processor.processSuccessfulCollection(new
MetricCollectionScope(AlertDomain.BUSINESS, "local",
java.util.Set.of("consumer.lag.total")),
List.of(current));
verify(alerts, never()).saveAlert(any(SystemAlertVO.class));
@@ -506,7 +511,7 @@ class NativeAlertProcessorTest {
new NativeAlertProcessor(service,
new NativeAlertEvaluationService(new AlertRuleEvaluator(), new
AlertStateMachine(), states,
mock(MetricSnapshotRepository.class), alerts, outbox,
suppression()),
- new AlertStateMachine(), states, alerts, outbox, suppression())
+ new AlertStateMachine(), states, alerts, outbox,
suppression(), mockTxManager())
.processSuccessfulCollection(new
MetricCollectionScope(AlertDomain.BUSINESS, "local",
java.util.Set.of("consumer.delay.seconds",
"consumer.lag.total")), List.of(lagSample));
@@ -536,12 +541,13 @@ class NativeAlertProcessorTest {
when(states.findActive(any(MetricCollectionScope.class),
eq(List.of(rule)))).thenReturn(List.of(active));
AlertRepository alerts = mock(AlertRepository.class);
- new NativeAlertProcessor(service,
+ NativeAlertProcessor processor = new NativeAlertProcessor(service,
new NativeAlertEvaluationService(new AlertRuleEvaluator(), new
AlertStateMachine(), states,
mock(MetricSnapshotRepository.class), alerts,
mock(NotificationOutboxService.class),
suppression()),
- new AlertStateMachine(), states, alerts,
mock(NotificationOutboxService.class), suppression())
- .processSuccessfulCollection(new
MetricCollectionScope(AlertDomain.BUSINESS, "local",
+ new AlertStateMachine(), states, alerts,
mock(NotificationOutboxService.class), suppression(),
+ mockTxManager());
+ processor.processSuccessfulCollection(new
MetricCollectionScope(AlertDomain.BUSINESS, "local",
java.util.Set.of("consumer.lag.total")), List.of(new
MetricSample("consumer.lag.total",
AlertDomain.BUSINESS, "local", null, Map.of(), null,
MetricAvailability.UNAVAILABLE,
Instant.now(), "BUSINESS_METRICS_COLLECTION_FAILED")));
@@ -550,6 +556,122 @@ class NativeAlertProcessorTest {
verify(alerts, never()).saveAlert(any(SystemAlertVO.class));
}
+ @Test
+ void singleLifecycleEmitFailureDoesNotRollBackTheBatchTest() {
+ AlertService service = mock(AlertService.class);
+ AlertRuleVO rule1 = rule(1L, "local", "orders", 1);
+ AlertRuleVO rule2 = rule(2L, "local", "payments", 1);
+
when(service.listRules(AlertDomain.BUSINESS)).thenReturn(List.of(rule1, rule2));
+ AlertStateRepository states = mock(AlertStateRepository.class);
+ when(states.save(any(AlertStateKey.class),
any(AlertRuleState.class))).thenReturn(true);
+ MetricSample ordersSample = sample("orders");
+ MetricSample paymentsSample = sample("payments");
+ ActiveAlertState active1 = new ActiveAlertState(
+ new AlertStateKey(1L, AlertFingerprint.of(1L, "local",
ordersSample.labels())),
+ new AlertRuleState(AlertStateStatus.FIRING, 1, 20D,
Instant.now().minusSeconds(60),
+ Instant.now().minusSeconds(60),
Instant.now().minusSeconds(60), null),
+ "local", ordersSample.labels());
+ ActiveAlertState active2 = new ActiveAlertState(
+ new AlertStateKey(2L, AlertFingerprint.of(2L, "local",
paymentsSample.labels())),
+ new AlertRuleState(AlertStateStatus.FIRING, 1, 20D,
Instant.now().minusSeconds(60),
+ Instant.now().minusSeconds(60),
Instant.now().minusSeconds(60), null),
+ "local", paymentsSample.labels());
+ when(states.findActive(any(MetricCollectionScope.class),
any())).thenReturn(List.of(active1, active2));
+ AlertRepository alerts = mock(AlertRepository.class);
+ when(alerts.saveAlert(any(SystemAlertVO.class)))
+ .thenThrow(new IllegalStateException("db write failed"))
+ .thenAnswer(invocation -> invocation.getArgument(0));
+ NotificationOutboxService outbox =
mock(NotificationOutboxService.class);
+
+ NativeAlertProcessor processor = new NativeAlertProcessor(service,
+ new NativeAlertEvaluationService(new AlertRuleEvaluator(), new
AlertStateMachine(), states,
+ mock(MetricSnapshotRepository.class), alerts, outbox,
suppression()),
+ new AlertStateMachine(), states, alerts, outbox,
suppression(), mockTxManager());
+ assertThatCode(() -> processor.processSuccessfulCollection(
+ new MetricCollectionScope(AlertDomain.BUSINESS, "local",
+ java.util.Set.of("consumer.lag.total")),
+ List.of())).doesNotThrowAnyException();
+
+ verify(outbox).enqueue(any(SystemAlertVO.class), eq(rule2), anyMap());
+ }
+
+ @Test
+ void emitFailureRollsBackBothSaveAndEmitPreventingOrphanEventsTest() {
+ AlertService service = mock(AlertService.class);
+ AlertRuleVO rule = rule("local", "orders", 1);
+
when(service.listRules(AlertDomain.BUSINESS)).thenReturn(List.of(rule));
+ MetricSample oldSample = sample("orders");
+ AlertStateKey oldKey = new AlertStateKey(rule.getId(),
+ AlertFingerprint.of(rule.getId(), oldSample.instanceId(),
oldSample.labels()));
+ ActiveAlertState active = new ActiveAlertState(oldKey,
+ new AlertRuleState(AlertStateStatus.FIRING, 1, 20D,
oldSample.collectedAt().minusSeconds(60),
+ oldSample.collectedAt().minusSeconds(60),
oldSample.collectedAt().minusSeconds(60), null),
+ oldSample.instanceId(), oldSample.labels());
+ AlertStateRepository states = mock(AlertStateRepository.class);
+ when(states.findActive(any(MetricCollectionScope.class),
eq(List.of(rule)))).thenReturn(List.of(active));
+ when(states.save(eq(oldKey),
any(AlertRuleState.class))).thenReturn(true);
+ AlertRepository alerts = mock(AlertRepository.class);
+ when(alerts.saveAlert(any(SystemAlertVO.class)))
+ .thenThrow(new IllegalStateException("event persist failed"));
+ NotificationOutboxService outbox =
mock(NotificationOutboxService.class);
+ PlatformTransactionManager txManager = mockTxManager();
+
+ NativeAlertProcessor processor = new NativeAlertProcessor(service,
+ new NativeAlertEvaluationService(new AlertRuleEvaluator(), new
AlertStateMachine(), states,
+ mock(MetricSnapshotRepository.class), alerts, outbox,
suppression()),
+ new AlertStateMachine(), states, alerts, outbox,
suppression(), txManager);
+ assertThatCode(() -> processor.processSuccessfulCollection(
+ new MetricCollectionScope(AlertDomain.BUSINESS, "local",
+ java.util.Set.of("consumer.lag.total")),
+ List.of())).doesNotThrowAnyException();
+
+ verify(states).save(eq(oldKey), any(AlertRuleState.class));
+ verify(alerts).saveAlert(any(SystemAlertVO.class));
+ verify(outbox, never()).enqueue(any(), any(), any());
+ verify(txManager).rollback(any(TransactionStatus.class));
+ }
+
+ @Test
+ void doesNotEmitResolvedEventWhenStateSaveLosesTheOptimisticRaceTest() {
+ AlertService service = mock(AlertService.class);
+ AlertRuleVO rule = rule("local", "orders", 1);
+
when(service.listRules(AlertDomain.BUSINESS)).thenReturn(List.of(rule));
+ MetricSample oldSample = sample("orders");
+ AlertStateKey oldKey = new AlertStateKey(rule.getId(),
+ AlertFingerprint.of(rule.getId(), oldSample.instanceId(),
oldSample.labels()));
+ ActiveAlertState active = new ActiveAlertState(oldKey,
+ new AlertRuleState(AlertStateStatus.FIRING, 1, 20D,
oldSample.collectedAt().minusSeconds(60),
+ oldSample.collectedAt().minusSeconds(60),
oldSample.collectedAt().minusSeconds(60), null),
+ oldSample.instanceId(), oldSample.labels());
+ AlertStateRepository states = mock(AlertStateRepository.class);
+ when(states.findActive(any(MetricCollectionScope.class),
eq(List.of(rule)))).thenReturn(List.of(active));
+ // a concurrent ACK already advanced the state, so this writer's save
loses the
+ // optimistic race and returns false
+ when(states.save(eq(oldKey),
any(AlertRuleState.class))).thenReturn(false);
+ AlertRepository alerts = mock(AlertRepository.class);
+ NotificationOutboxService outbox =
mock(NotificationOutboxService.class);
+
+ NativeAlertProcessor processor = new NativeAlertProcessor(service,
+ new NativeAlertEvaluationService(new AlertRuleEvaluator(), new
AlertStateMachine(), states,
+ mock(MetricSnapshotRepository.class), alerts, outbox,
suppression()),
+ new AlertStateMachine(), states, alerts, outbox,
suppression(), mockTxManager());
+ assertThatCode(() -> processor.processSuccessfulCollection(
+ new MetricCollectionScope(AlertDomain.BUSINESS, "local",
+ java.util.Set.of("consumer.lag.total")),
+ List.of())).doesNotThrowAnyException();
+
+ verify(states).save(eq(oldKey), any(AlertRuleState.class));
+ verify(alerts, never()).saveAlert(any(SystemAlertVO.class));
+ verify(outbox, never()).enqueue(any(), any(), any());
+ }
+
+ private static PlatformTransactionManager mockTxManager() {
+ PlatformTransactionManager txManager =
mock(PlatformTransactionManager.class);
+ TransactionStatus status = mock(TransactionStatus.class);
+
when(txManager.getTransaction(any(TransactionDefinition.class))).thenReturn(status);
+ return txManager;
+ }
+
private static AlertNotificationSuppressionService suppression() {
AlertNotificationSuppressionService service =
mock(AlertNotificationSuppressionService.class);
when(service.findSuppressingClusterAlert(any(SystemAlertVO.class))).thenReturn(Optional.empty());
