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 66519d03d fix(alert,deploy): fence stale rule evaluations; align
upgrade schema (#6312)
66519d03d is described below
commit 66519d03d21393fcde0cc8298ecce31be2147333
Author: pilichoumao <[email protected]>
AuthorDate: Wed Oct 28 00:38:55 2026 +0800
fix(alert,deploy): fence stale rule evaluations; align upgrade schema
(#6312)
* fix(alert): reject stale rule evaluations under a row lock (#6312)
Evaluation sampled a rule, spent time collecting and aggregating, then wrote
runtime state, lifecycle events and notification outbox rows against that
sample -
so a rule that was disabled or reconfigured in between still got the stale
write.
Both the sampled-evaluation and the missing-state reconciliation paths now
take a
current SELECT ... FOR UPDATE read of the rule inside their existing
transaction
(findRuleByIdForUpdate) and skip the write when the rule is gone, disabled,
or its
evaluation identity differs from the snapshot. The lock is held through the
writes,
so a concurrent management update waits and then applies its normal
runtime-state
cleanup. A locking read also observes committed management changes under
MySQL
REPEATABLE READ, which a second ordinary snapshot read would not. The
fingerprint
accessor is renamed of -> evaluationIdentity to match AlertService.
The test's timing assertion was replaced with a positive observation of the
lock
wait (information_schema.innodb_trx under MySQL,
INFORMATION_SCHEMA.SESSIONS under
H2) so it cannot pass vacuously on a slow CI runner, and twelve duplicated
stubCurrentRules calls were dropped in favour of the shared helper.
* fix(deploy): align cloud credential upgrade schema (#6210)
deploy/mysql/upgrade-cloud-vendor.sql created the credential table with
`id VARCHAR(64)`, `created_at` / `updated_at` and a `credential_id
VARCHAR(64)`
foreign key, while schema.sql - and the entity, which declares
`@TableId(type = AUTO) Long id` with gmtCreate / gmtModified - uses a bigint
auto-increment primary key and the gmt_* columns. A database built by the
upgrade
script therefore did not match the one built from schema.sql, and
CloudCredentialSchemaMigration only adds indexes and foreign keys, so
nothing
corrected the columns afterwards. Align the script with schema.sql column by
column.
Batch integration of two fixes by the same contributor; #6210 is closed as
merged
into this one. Verified: 237 backend tests across the affected classes,
AlertRuleEvaluationConcurrencyTest 6/6 on three consecutive runs, and
dropping
FOR UPDATE from the locking read fails
disableShouldWaitForLockedEvaluationAndThenPurgeItsStateTest - which is
what shows
the rewritten lock-wait assertion is really observing the lock.
---
deploy/mysql/upgrade-cloud-vendor.sql | 10 +-
docs/studio-native-alerting-design.md | 2 +
.../rocketmq/studio/ops/alert/AlertRepository.java | 3 +
.../ops/alert/MybatisPlusAlertRepository.java | 11 +
.../ops/alert/NativeAlertEvaluationService.java | 7 +
.../studio/ops/alert/NativeAlertProcessor.java | 6 +-
.../studio/ops/alert/NativeAlertRulePolicy.java | 9 +
.../alert/AlertRuleEvaluationConcurrencyTest.java | 259 +++++++++++++++++++++
.../alert/NativeAlertClusterCorrelationTest.java | 3 +
.../NativeAlertEvaluationTransactionTest.java | 1 +
.../studio/ops/alert/NativeAlertProcessorTest.java | 23 +-
.../persistence/DemoDataSqlCompatibilityTest.java | 52 +++++
12 files changed, 379 insertions(+), 7 deletions(-)
diff --git a/deploy/mysql/upgrade-cloud-vendor.sql
b/deploy/mysql/upgrade-cloud-vendor.sql
index ca7ac15e9..5dc6646ba 100644
--- a/deploy/mysql/upgrade-cloud-vendor.sql
+++ b/deploy/mysql/upgrade-cloud-vendor.sql
@@ -2,6 +2,7 @@
-- 存量 MySQL 数据卷增量迁移：rmq_instance 增加厂商维度 + 新增 rmq_cloud_credential 凭据表
-- 适用：数据卷已初始化、docker-entrypoint-initdb.d 不会再执行的存量部署。
-- 全新数据卷由 server/src/main/resources/db/schema.sql 直接覆盖，无需本脚本。
+-- 仅补充缺失的列和表；不会转换已有的 UUID 主键、凭据引用或旧时间列。
-- 幂等：可重复执行。
--
-- 用法（远程容器内执行）：
@@ -51,7 +52,7 @@ SET @credential_id_column_exists := (
AND column_name = 'credential_id'
);
SET @credential_id_sql := IF(@credential_id_column_exists = 0,
- "ALTER TABLE rmq_instance ADD COLUMN credential_id VARCHAR(64) COMMENT '引用
rmq_cloud_credential.id' AFTER cloud_instance_id",
+ "ALTER TABLE rmq_instance ADD COLUMN credential_id bigint(20) unsigned
COMMENT '引用 rmq_cloud_credential.id' AFTER cloud_instance_id",
'SELECT 1');
PREPARE credential_id_statement FROM @credential_id_sql;
EXECUTE credential_id_statement;
@@ -93,13 +94,14 @@ UPDATE rmq_instance SET vendor = 'APACHE' WHERE vendor IS
NULL OR vendor = '';
-- 7. 云厂商凭据表（secret_key 为 base64 编码，禁止明文；access_key 明文用于唯一键与打码展示）
CREATE TABLE IF NOT EXISTS rmq_cloud_credential (
- id VARCHAR(64) PRIMARY KEY,
+ `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT '主键',
+ `gmt_create` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
+ `gmt_modified` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE
CURRENT_TIMESTAMP COMMENT '修改时间',
name VARCHAR(128) NOT NULL COMMENT '凭据显示名',
vendor VARCHAR(32) NOT NULL COMMENT 'ALIYUN/TENCENT',
access_key VARCHAR(255) NOT NULL,
secret_key VARCHAR(512) NOT NULL COMMENT 'base64 编码的 SK',
remark VARCHAR(255),
- created_at DATETIME DEFAULT CURRENT_TIMESTAMP,
- updated_at DATETIME DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+ PRIMARY KEY (`id`),
UNIQUE KEY uk_vendor_access_key (vendor, access_key)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
diff --git a/docs/studio-native-alerting-design.md
b/docs/studio-native-alerting-design.md
index 963a40df6..bbddc7004 100644
--- a/docs/studio-native-alerting-design.md
+++ b/docs/studio-native-alerting-design.md
@@ -191,6 +191,8 @@ consecutiveSamples=2, severity=CRITICAL
The evaluator runs after a collector writes a new sample. It computes an
aggregation over the bounded local snapshot window, then advances an alert
state keyed by a stable fingerprint.
+Before writing runtime state, lifecycle events, or notification outbox rows,
both sampled evaluation and missing-state reconciliation take a current `SELECT
... FOR UPDATE` read of the rule inside their existing transaction. A deleted
or disabled rule, or a rule whose evaluation conditions differ from the
collection snapshot, is skipped. The lock remains held through those writes, so
a concurrent management update waits and then applies its normal runtime-state
cleanup. A current locking [...]
+
```text
OK -- condition met --> PENDING -- consecutive samples met --> FIRING
FIRING -- condition clear --> RESOLVED
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertRepository.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertRepository.java
index 79ac05ad7..8e9ea3f12 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertRepository.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/AlertRepository.java
@@ -32,6 +32,9 @@ public interface AlertRepository {
Optional<AlertRuleVO> findRuleById(Long id);
+ /** Current locking read; the caller must hold a transaction through all
lifecycle writes. */
+ Optional<AlertRuleVO> findRuleByIdForUpdate(Long id);
+
List<AlertRuleVO> findRulesByIds(List<Long> ids);
AlertRuleVO insertRule(AlertRuleVO rule);
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/MybatisPlusAlertRepository.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/MybatisPlusAlertRepository.java
index 1c6e7e13d..607327a13 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/MybatisPlusAlertRepository.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/MybatisPlusAlertRepository.java
@@ -31,6 +31,7 @@ import
org.apache.rocketmq.studio.persistence.mapper.RmqAlertNotificationOutboxM
import org.apache.rocketmq.studio.persistence.mapper.RmqSystemAlertMapper;
import org.springframework.stereotype.Repository;
import org.springframework.transaction.annotation.Transactional;
+import org.springframework.transaction.annotation.Propagation;
import org.springframework.util.StringUtils;
import lombok.RequiredArgsConstructor;
@@ -119,6 +120,16 @@ public class MybatisPlusAlertRepository implements
AlertRepository {
.map(MybatisPlusAlertRepository::toRuleVO);
}
+ @Override
+ @Transactional(propagation = Propagation.MANDATORY)
+ public Optional<AlertRuleVO> findRuleByIdForUpdate(Long id) {
+ if (id == null) {
+ return Optional.empty();
+ }
+ return Optional.ofNullable(ruleMapper.selectOne(new
QueryWrapper<RmqAlertRule>()
+ .eq("id", id).last("FOR
UPDATE"))).map(MybatisPlusAlertRepository::toRuleVO);
+ }
+
@Override
public List<AlertRuleVO> findRulesByIds(List<Long> ids) {
if (ids == null || ids.isEmpty()) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NativeAlertEvaluationService.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NativeAlertEvaluationService.java
index fd490ea5a..a66db867a 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NativeAlertEvaluationService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NativeAlertEvaluationService.java
@@ -50,6 +50,13 @@ public class NativeAlertEvaluationService {
@Transactional(propagation = Propagation.REQUIRES_NEW)
public void evaluate(AlertRuleVO rule, MetricSample sample) {
MetricSample evaluatedSample =
NativeAlertSampleAggregator.aggregate(rule, sample, snapshotRepository);
+ // Aggregation can block while management commits. A current locking
read, rather than
+ // another snapshot read, also sees those commits under MySQL
REPEATABLE READ.
+ Optional<AlertRuleVO> current =
NativeAlertRulePolicy.lockMatchingRule(alertRepository, rule);
+ if (current.isEmpty()) {
+ return;
+ }
+ rule = current.get();
AlertEvaluationResult evaluation = evaluator.evaluate(rule,
evaluatedSample);
if (!evaluation.matches()) {
return;
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 e8d8ace88..dd32d59c1 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
@@ -172,10 +172,14 @@ public class NativeAlertProcessor {
// emit is gated on the save result: an alert that already
moved on must
// not receive a RESOLVED event.
isolatedTx.executeWithoutResult(txStatus -> {
+ Optional<AlertRuleVO> current =
NativeAlertRulePolicy.lockMatchingRule(alertRepository, rule);
+ if (current.isEmpty()) {
+ return;
+ }
if (!stateRepository.save(active.key(), update.state())) {
return;
}
- emitLifecycleEvent(rule, active.key(), update,
scope.domain(),
+ emitLifecycleEvent(current.get(), active.key(), update,
scope.domain(),
active.instanceId(), rule.getMetric(),
active.labels(), resolvedAt);
});
} catch (RuntimeException error) {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NativeAlertRulePolicy.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NativeAlertRulePolicy.java
index b8a1085be..9e419f874 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NativeAlertRulePolicy.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NativeAlertRulePolicy.java
@@ -22,6 +22,7 @@ import org.springframework.util.StringUtils;
import java.util.Locale;
import java.util.Map;
import java.util.Set;
+import java.util.Optional;
/** Enforces the scope required to safely evaluate Studio-native metric rules.
*/
final class NativeAlertRulePolicy {
@@ -50,6 +51,14 @@ final class NativeAlertRulePolicy {
private NativeAlertRulePolicy() {
}
+ /** Lock before writing, then reject the old collection snapshot if its
rule no longer applies. */
+ static Optional<AlertRuleVO> lockMatchingRule(AlertRepository repository,
AlertRuleVO snapshot) {
+ return repository.findRuleByIdForUpdate(snapshot.getId())
+ .filter(AlertRuleVO::isEnabled)
+ .filter(current ->
AlertRuleSemanticFingerprint.evaluationIdentity(current)
+
.equals(AlertRuleSemanticFingerprint.evaluationIdentity(snapshot)));
+ }
+
static boolean isNativeMetric(String metric) {
return StringUtils.hasText(metric) &&
NATIVE_METRICS.containsKey(metric.trim());
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertRuleEvaluationConcurrencyTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertRuleEvaluationConcurrencyTest.java
new file mode 100644
index 000000000..e64585ba9
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/AlertRuleEvaluationConcurrencyTest.java
@@ -0,0 +1,259 @@
+/*
+ * 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 java.time.Instant;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
+import org.apache.rocketmq.studio.cluster.metrics.MetricAvailability;
+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.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+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.bean.override.mockito.MockitoBean;
+import org.springframework.test.context.bean.override.mockito.MockitoSpyBean;
+import org.springframework.test.util.AopTestUtils;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.transaction.support.TransactionTemplate;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.when;
+
+/** Runs with the CI MySQL datasource, or with the dev/H2 profile for local
verification. */
+@SpringBootTest(properties =
"studio.alerting.notification-dispatch-interval=PT24H")
+class AlertRuleEvaluationConcurrencyTest {
+ @Autowired
+ private AlertService alerts;
+ @Autowired
+ private NativeAlertEvaluationService evaluation;
+ @Autowired
+ private NativeAlertProcessor processor;
+ @Autowired
+ private JdbcTemplate jdbc;
+ @Autowired
+ private PlatformTransactionManager transactionManager;
+ @MockitoBean
+ private MetricSnapshotRepository snapshots;
+ @MockitoSpyBean
+ private AlertRepository repository;
+ @MockitoSpyBean
+ private AlertStateRepository states;
+ private AlertRuleVO rule;
+ private MetricSample sample;
+
+ @BeforeEach
+ void initializeTest() {
+ String group = "race-" + UUID.randomUUID();
+ rule =
alerts.createRule(AlertRuleVO.builder().name(group).domain(AlertDomain.BUSINESS)
+
.metric("consumer.lag.total").operator(">").threshold(10).instanceId("probe-instance")
+
.consumerGroup(group).enabled(true).windowSeconds(60).channels(List.of("email")).build());
+ sample = new MetricSample(rule.getMetric(), AlertDomain.BUSINESS,
rule.getInstanceId(), null,
+ Map.of("consumerGroup", group), 20D,
MetricAvailability.AVAILABLE, Instant.now());
+ when(snapshots.findRecent(any(), any())).thenReturn(List.of(sample));
+ }
+
+ @AfterEach
+ void cleanupTest() {
+ if (rule == null) {
+ return;
+ }
+ jdbc.update("DELETE FROM rmq_alert_notification_outbox WHERE alert_id
IN "
+ + "(SELECT id FROM rmq_system_alert WHERE rule_id=?)",
rule.getId());
+ jdbc.update("DELETE FROM rmq_system_alert WHERE rule_id=?",
rule.getId());
+ jdbc.update("DELETE FROM rmq_alert_state WHERE rule_id=?",
rule.getId());
+ jdbc.update("DELETE FROM rmq_alert_rule WHERE id=?", rule.getId());
+ }
+
+ private String databaseProduct() {
+ return
jdbc.execute((org.springframework.jdbc.core.ConnectionCallback<String>)
connection ->
+ connection.getMetaData().getDatabaseProductName());
+ }
+
+ private long sessionId() {
+ return jdbc.queryForObject("MySQL".equals(databaseProduct())
+ ? "SELECT CONNECTION_ID()" : "SELECT SESSION_ID()",
Long.class);
+ }
+
+ /**
+ * Observes the management write actually waiting on the evaluation's row
lock, instead of
+ * inferring it from "still unfinished after a fixed delay" - a slow
runner would pass such an
+ * assertion without ever reaching the statement. MySQL reports the wait in
+ * information_schema.innodb_trx, the dev/H2 profile in
INFORMATION_SCHEMA.SESSIONS, so the
+ * check stays deterministic on both datasources and is skipped on neither.
+ */
+ private boolean awaitManagementLockWait(long timeoutMillis) throws
InterruptedException {
+ String waitingSessions = "MySQL".equals(databaseProduct())
+ ? "SELECT COUNT(*) FROM information_schema.innodb_trx WHERE
trx_state = 'LOCK WAIT'"
+ : "SELECT COUNT(*) FROM INFORMATION_SCHEMA.SESSIONS WHERE
SESSION_STATE = 'BLOCKED'";
+ long deadline = System.nanoTime() +
TimeUnit.MILLISECONDS.toNanos(timeoutMillis);
+ while (System.nanoTime() < deadline) {
+ Integer waiting = jdbc.queryForObject(waitingSessions,
Integer.class);
+ if (waiting != null && waiting > 0) {
+ return true;
+ }
+ Thread.sleep(5);
+ }
+ return false;
+ }
+
+ private void assertRows(int expectedStates, int expectedEvents) {
+ assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM rmq_alert_state
WHERE rule_id=?",
+ Integer.class, rule.getId())).isEqualTo(expectedStates);
+ assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM rmq_system_alert
WHERE rule_id=?",
+ Integer.class, rule.getId())).isEqualTo(expectedEvents);
+ assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM
rmq_alert_notification_outbox WHERE alert_id IN "
+ + "(SELECT id FROM rmq_system_alert WHERE rule_id=?)",
Integer.class, rule.getId()))
+ .isEqualTo(expectedEvents);
+ }
+
+ private void afterPausedEvaluation(Consumer<AlertRuleVO> mutation, int
expectedRows) throws Exception {
+ CountDownLatch started = new CountDownLatch(1);
+ CountDownLatch resume = new CountDownLatch(1);
+ AtomicReference<Long> evaluationSession = new AtomicReference<>();
+ when(snapshots.findRecent(any(), any())).thenAnswer(inv -> {
+ evaluationSession.set(sessionId());
+ // Establish a real MVCC snapshot before the management
transaction commits.
+
assertThat(repository.findRuleById(rule.getId()).orElseThrow().isEnabled()).isTrue();
+ started.countDown();
+ assertThat(resume.await(10, TimeUnit.SECONDS)).isTrue();
+ return List.of(sample);
+ });
+ try (var executor = Executors.newSingleThreadExecutor()) {
+ var pending = executor.submit(() -> evaluation.evaluate(rule,
sample));
+ try {
+ assertThat(started.await(10, TimeUnit.SECONDS)).isTrue();
+ Long controlSession = new
TransactionTemplate(transactionManager).execute(status -> {
+ long id = sessionId();
+
mutation.accept(repository.findRuleById(rule.getId()).orElseThrow());
+ return id;
+ });
+
assertThat(controlSession).isNotEqualTo(evaluationSession.get());
+ resume.countDown();
+ pending.get(10, TimeUnit.SECONDS);
+ assertRows(expectedRows, expectedRows);
+ } finally {
+ resume.countDown();
+ }
+ }
+ }
+
+ @Test
+ void committedDisableShouldRejectOldEvaluationTest() throws Exception {
+ afterPausedEvaluation(current ->
alerts.toggleRule(AlertDomain.BUSINESS, current.getId(), false), 0);
+
assertThat(repository.findRuleById(rule.getId()).orElseThrow().isEnabled()).isFalse();
+ }
+
+ @Test
+ void committedDeleteShouldRejectOldEvaluationTest() throws Exception {
+ afterPausedEvaluation(current ->
alerts.deleteRule(AlertDomain.BUSINESS, current.getId()), 0);
+ assertThat(repository.findRuleById(rule.getId())).isEmpty();
+ }
+
+ @Test
+ void changedConditionsShouldRejectOldEvaluationTest() throws Exception {
+ afterPausedEvaluation(current -> {
+ current.setThreshold(100);
+ alerts.updateRule(AlertDomain.BUSINESS, current);
+ }, 0);
+ }
+
+ @Test
+ void presentationEditShouldUseCurrentNotificationFieldsTest() throws
Exception {
+ afterPausedEvaluation(current -> {
+ current.setName("Updated presentation");
+ current.setNotificationTemplate("updated-template");
+ alerts.updateRule(AlertDomain.BUSINESS, current);
+ }, 1);
+ assertThat(jdbc.queryForObject("SELECT title FROM rmq_system_alert
WHERE rule_id=?",
+ String.class, rule.getId())).isEqualTo("Updated presentation");
+ assertThat(jdbc.queryForObject("SELECT message_content FROM
rmq_alert_notification_outbox WHERE alert_id IN "
+ + "(SELECT id FROM rmq_system_alert WHERE rule_id=?)",
String.class, rule.getId()))
+ .isEqualTo("updated-template");
+ }
+
+ @Test
+ void disableShouldWaitForLockedEvaluationAndThenPurgeItsStateTest() throws
Exception {
+ CountDownLatch locked = new CountDownLatch(1);
+ CountDownLatch resume = new CountDownLatch(1);
+ CountDownLatch managementStarted = new CountDownLatch(1);
+ doAnswer(inv -> {
+ Object current = inv.callRealMethod();
+ locked.countDown();
+ assertThat(resume.await(10, TimeUnit.SECONDS)).isTrue();
+ return current;
+ }).when((AlertRepository)
AopTestUtils.getUltimateTargetObject(repository))
+ .findRuleByIdForUpdate(rule.getId());
+ try (var executor = Executors.newFixedThreadPool(2)) {
+ var pending = executor.submit(() -> evaluation.evaluate(rule,
sample));
+ try {
+ assertThat(locked.await(10, TimeUnit.SECONDS)).isTrue();
+ var disable = executor.submit(() -> {
+ managementStarted.countDown();
+ alerts.toggleRule(AlertDomain.BUSINESS, rule.getId(),
false);
+ });
+ assertThat(managementStarted.await(10,
TimeUnit.SECONDS)).isTrue();
+
assertThat(awaitManagementLockWait(TimeUnit.SECONDS.toMillis(10))).isTrue();
+ resume.countDown();
+ pending.get(10, TimeUnit.SECONDS);
+ disable.get(10, TimeUnit.SECONDS);
+ assertRows(0, 1); // Existing events/outbox are history;
disabling only purges runtime state.
+
assertThat(repository.findRuleById(rule.getId()).orElseThrow().isEnabled()).isFalse();
+ } finally {
+ resume.countDown();
+ }
+ }
+ }
+
+ @Test
+ void committedDisableShouldRejectStaleMissingStateReconciliationTest()
throws Exception {
+ evaluation.evaluate(rule, sample);
+ assertRows(1, 1);
+ CountDownLatch read = new CountDownLatch(1);
+ CountDownLatch resume = new CountDownLatch(1);
+ doAnswer(inv -> {
+ Object active = inv.callRealMethod();
+ read.countDown();
+ assertThat(resume.await(10, TimeUnit.SECONDS)).isTrue();
+ return active;
+ }).when((AlertStateRepository)
AopTestUtils.getUltimateTargetObject(states)).findActive(any(), any());
+ try (var executor = Executors.newSingleThreadExecutor()) {
+ var pending = executor.submit(() ->
processor.processSuccessfulCollection(new MetricCollectionScope(
+ AlertDomain.BUSINESS, rule.getInstanceId(),
java.util.Set.of(rule.getMetric())), List.of()));
+ try {
+ assertThat(read.await(10, TimeUnit.SECONDS)).isTrue();
+ alerts.toggleRule(AlertDomain.BUSINESS, rule.getId(), false);
+ resume.countDown();
+ pending.get(10, TimeUnit.SECONDS);
+ assertRows(0, 1);
+ } finally {
+ resume.countDown();
+ }
+ }
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertClusterCorrelationTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertClusterCorrelationTest.java
index 5d590afe5..945a73d73 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertClusterCorrelationTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertClusterCorrelationTest.java
@@ -67,6 +67,7 @@ class NativeAlertClusterCorrelationTest {
when(states.find(any())).thenReturn(Optional.empty());
when(states.save(any(), any())).thenReturn(true);
AlertRepository alerts = mock(AlertRepository.class);
+
when(alerts.findRuleByIdForUpdate(rule.getId())).thenReturn(Optional.of(rule));
when(alerts.saveAlert(any())).thenAnswer(invocation ->
invocation.getArgument(0));
NotificationOutboxService outbox =
mock(NotificationOutboxService.class);
AlertNotificationSuppressionService suppression =
mock(AlertNotificationSuppressionService.class);
@@ -143,6 +144,8 @@ class NativeAlertClusterCorrelationTest {
when(states.save(any(), any())).thenReturn(true);
List<SystemAlertVO> emitted = new ArrayList<>();
AlertRepository alerts = mock(AlertRepository.class);
+
when(alerts.findRuleByIdForUpdate(clusterRule.getId())).thenReturn(Optional.of(clusterRule));
+
when(alerts.findRuleByIdForUpdate(businessRule.getId())).thenReturn(Optional.of(businessRule));
when(alerts.saveAlert(any())).thenAnswer(invocation -> {
SystemAlertVO event = invocation.getArgument(0);
event.setId((long) emitted.size() + 1);
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertEvaluationTransactionTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertEvaluationTransactionTest.java
index 35a770281..6af0b6f77 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertEvaluationTransactionTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertEvaluationTransactionTest.java
@@ -86,6 +86,7 @@ class NativeAlertEvaluationTransactionTest {
AlertRuleVO rule =
AlertRuleVO.builder().id(1L).domain(AlertDomain.BUSINESS).name("Orders")
.metric("consumer.lag.total").operator(">").threshold(10).enabled(true).instanceId("local")
.consumerGroup("orders").build();
+
when(alertRepository.findRuleByIdForUpdate(rule.getId())).thenReturn(Optional.of(rule));
MetricSample sample = new MetricSample("consumer.lag.total",
AlertDomain.BUSINESS, "local", null,
Map.of("consumerGroup", "orders"), 20D,
MetricAvailability.AVAILABLE, Instant.now());
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 ce0e77e09..9ff55fed2 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
@@ -172,7 +172,7 @@ class NativeAlertProcessorTest {
processor.process(List.of(sample("orders")));
assertThat(saved.values()).singleElement().extracting(AlertRuleState::status)
.isEqualTo(AlertStateStatus.PENDING);
- verifyNoInteractions(alerts);
+ verify(alerts, never()).saveAlert(any(SystemAlertVO.class));
processor.process(List.of(sample("orders")));
assertThat(saved.values()).singleElement().extracting(AlertRuleState::status)
@@ -271,7 +271,7 @@ class NativeAlertProcessorTest {
processor(service, states, alerts).process(List.of(sample("orders")));
- verifyNoInteractions(alerts);
+ verify(alerts, never()).saveAlert(any(SystemAlertVO.class));
}
@Test
@@ -432,9 +432,18 @@ class NativeAlertProcessorTest {
org.mockito.ArgumentMatchers.anyMap());
}
+ private static void stubCurrentRules(AlertService service, AlertRepository
repository) {
+
when(repository.findRuleByIdForUpdate(any(Long.class))).thenAnswer(invocation ->
+
java.util.stream.Stream.concat(service.listRules(AlertDomain.BUSINESS).stream(),
+
service.listRules(AlertDomain.CLUSTER).stream())
+ .filter(rule -> java.util.Objects.equals(rule.getId(),
invocation.getArgument(0)))
+ .findFirst());
+ }
+
private static NativeAlertProcessor processor(AlertService service,
AlertStateRepository states,
MetricSnapshotRepository snapshots, AlertRepository alerts,
NotificationOutboxService outbox,
AlertNotificationSuppressionService suppression) {
+ stubCurrentRules(service, alerts);
NativeAlertEvaluationService evaluationService = new
NativeAlertEvaluationService(new AlertRuleEvaluator(),
new AlertStateMachine(), states, snapshots, alerts, outbox,
suppression);
return new NativeAlertProcessor(service, evaluationService, new
AlertStateMachine(), states, alerts, outbox,
@@ -512,6 +521,7 @@ class NativeAlertProcessorTest {
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);
+ stubCurrentRules(service, alerts);
when(alerts.saveAlert(any(SystemAlertVO.class))).thenAnswer(invocation
-> invocation.getArgument(0));
NotificationOutboxService outbox =
mock(NotificationOutboxService.class);
@@ -553,6 +563,7 @@ class NativeAlertProcessorTest {
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);
+ stubCurrentRules(service, alerts);
when(alerts.saveAlert(any(SystemAlertVO.class))).thenAnswer(invocation
-> invocation.getArgument(0));
NotificationOutboxService outbox =
mock(NotificationOutboxService.class);
@@ -593,6 +604,7 @@ class NativeAlertProcessorTest {
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);
+ stubCurrentRules(service, alerts);
when(alerts.saveAlert(any(SystemAlertVO.class))).thenAnswer(invocation
-> invocation.getArgument(0));
NotificationOutboxService outbox =
mock(NotificationOutboxService.class);
@@ -626,6 +638,7 @@ class NativeAlertProcessorTest {
when(states.save(eq(key), any(AlertRuleState.class))).thenReturn(true);
when(states.findActive(any(MetricCollectionScope.class),
eq(List.of(rule)))).thenReturn(List.of(active));
AlertRepository alerts = mock(AlertRepository.class);
+ stubCurrentRules(service, alerts);
NativeAlertProcessor processor = new NativeAlertProcessor(service,
new NativeAlertEvaluationService(new AlertRuleEvaluator(), new
AlertStateMachine(), states,
@@ -659,6 +672,7 @@ class NativeAlertProcessorTest {
when(states.save(eq(key), any(AlertRuleState.class))).thenReturn(true);
when(states.findActive(any(MetricCollectionScope.class),
eq(List.of(rule)))).thenReturn(List.of(active));
AlertRepository alerts = mock(AlertRepository.class);
+ stubCurrentRules(service, alerts);
NotificationOutboxService outbox =
mock(NotificationOutboxService.class);
NativeAlertProcessor processor = new NativeAlertProcessor(service,
@@ -692,6 +706,7 @@ class NativeAlertProcessorTest {
when(states.findActive(any(MetricCollectionScope.class),
eq(List.of(rule)))).thenReturn(List.of(active));
when(states.save(eq(key), any(AlertRuleState.class))).thenReturn(true);
AlertRepository alerts = mock(AlertRepository.class);
+ stubCurrentRules(service, alerts);
when(alerts.saveAlert(any(SystemAlertVO.class))).thenAnswer(invocation
-> invocation.getArgument(0));
NotificationOutboxService outbox =
mock(NotificationOutboxService.class);
@@ -730,6 +745,7 @@ class NativeAlertProcessorTest {
AlertStateRepository states = mock(AlertStateRepository.class);
when(states.findActive(any(MetricCollectionScope.class),
eq(List.of(rule)))).thenReturn(List.of(active));
AlertRepository alerts = mock(AlertRepository.class);
+ stubCurrentRules(service, alerts);
NativeAlertProcessor processor = new NativeAlertProcessor(service,
new NativeAlertEvaluationService(new AlertRuleEvaluator(), new
AlertStateMachine(), states,
@@ -899,6 +915,7 @@ class NativeAlertProcessorTest {
"local", paymentsSample.labels());
when(states.findActive(any(MetricCollectionScope.class),
any())).thenReturn(List.of(active1, active2));
AlertRepository alerts = mock(AlertRepository.class);
+ stubCurrentRules(service, alerts);
when(alerts.saveAlert(any(SystemAlertVO.class)))
.thenThrow(new IllegalStateException("db write failed"))
.thenAnswer(invocation -> invocation.getArgument(0));
@@ -932,6 +949,7 @@ class NativeAlertProcessorTest {
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);
+ stubCurrentRules(service, alerts);
when(alerts.saveAlert(any(SystemAlertVO.class)))
.thenThrow(new IllegalStateException("event persist failed"));
NotificationOutboxService outbox =
mock(NotificationOutboxService.class);
@@ -970,6 +988,7 @@ class NativeAlertProcessorTest {
// optimistic race and returns false
when(states.save(eq(oldKey),
any(AlertRuleState.class))).thenReturn(false);
AlertRepository alerts = mock(AlertRepository.class);
+ stubCurrentRules(service, alerts);
NotificationOutboxService outbox =
mock(NotificationOutboxService.class);
NativeAlertProcessor processor = new NativeAlertProcessor(service,
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/persistence/DemoDataSqlCompatibilityTest.java
b/server/src/test/java/org/apache/rocketmq/studio/persistence/DemoDataSqlCompatibilityTest.java
index 2f43f2376..f598ba5ac 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/persistence/DemoDataSqlCompatibilityTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/persistence/DemoDataSqlCompatibilityTest.java
@@ -30,6 +30,9 @@ import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
+import java.sql.Types;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -87,6 +90,55 @@ class DemoDataSqlCompatibilityTest {
}
}
+ @Test
+ void cloudVendorUpgradeCreatesUsableNumericCredentialsTest() throws
Exception {
+ String script =
Files.readString(resolveDeployScript("upgrade-cloud-vendor.sql"),
StandardCharsets.UTF_8);
+ // H2 cannot execute the MySQL PREPARE/EXECUTE guards. Exercise the
actual DDL
+ // selected by those guards; full-script guard coverage requires MySQL.
+ Matcher credentialColumn = Pattern.compile("\"(ALTER TABLE
rmq_instance ADD COLUMN credential_id [^\"]+)\"")
+ .matcher(script);
+ assertThat(credentialColumn.find()).isTrue();
+ String createCredentials = script.substring(script.indexOf("CREATE
TABLE IF NOT EXISTS rmq_cloud_credential"));
+ try (Connection connection = DriverManager.getConnection(
+
"jdbc:h2:mem:cloud-vendor-sql;MODE=MySQL;DATABASE_TO_LOWER=TRUE", "sa", "");
+ Statement statement = connection.createStatement()) {
+ statement.executeUpdate("CREATE TABLE rmq_instance (id BIGINT
AUTO_INCREMENT PRIMARY KEY,"
+ + " cloud_instance_id VARCHAR(128))");
+ statement.executeUpdate(credentialColumn.group(1));
+ ScriptUtils.executeSqlScript(connection, new ByteArrayResource(
+ createCredentials.getBytes(StandardCharsets.UTF_8), "cloud
credential DDL"));
+
+ assertThat(hasColumn(connection, "rmq_cloud_credential",
"gmt_create")).isTrue();
+ assertThat(hasColumn(connection, "rmq_cloud_credential",
"gmt_modified")).isTrue();
+ assertThat(hasColumn(connection, "rmq_cloud_credential",
"created_at")).isFalse();
+ try (ResultSet columns = connection.getMetaData().getColumns(null,
null, "rmq_cloud_credential", "id")) {
+ assertThat(columns.next()).isTrue();
+
assertThat(columns.getInt("DATA_TYPE")).isEqualTo(Types.BIGINT);
+
assertThat(columns.getString("IS_AUTOINCREMENT")).isEqualTo("YES");
+ }
+ try (ResultSet columns = connection.getMetaData().getColumns(null,
null, "rmq_instance", "credential_id")) {
+ assertThat(columns.next()).isTrue();
+
assertThat(columns.getInt("DATA_TYPE")).isEqualTo(Types.BIGINT);
+ }
+ statement.executeUpdate("INSERT INTO rmq_cloud_credential (name,
vendor, access_key, secret_key)"
+ + " VALUES ('test-credential', 'ALIYUN',
'test-access-key', 'dGVzdA==')");
+ statement.executeUpdate("INSERT INTO rmq_instance (credential_id)"
+ + " SELECT id FROM rmq_cloud_credential WHERE name =
'test-credential'");
+ statement.executeUpdate("ALTER TABLE rmq_instance ADD CONSTRAINT
fk_test_credential"
+ + " FOREIGN KEY (credential_id) REFERENCES
rmq_cloud_credential(id)");
+ ScriptUtils.executeSqlScript(connection, new ByteArrayResource(
+ createCredentials.getBytes(StandardCharsets.UTF_8),
"repeated cloud credential DDL"));
+ try (ResultSet rows = statement.executeQuery("SELECT c.id,
c.gmt_create, c.gmt_modified"
+ + " FROM rmq_cloud_credential c JOIN rmq_instance i ON
i.credential_id = c.id")) {
+ assertThat(rows.next()).isTrue();
+ assertThat(rows.getLong("id")).isPositive();
+ assertThat(rows.getTimestamp("gmt_create")).isNotNull();
+ assertThat(rows.getTimestamp("gmt_modified")).isNotNull();
+ assertThat(rows.next()).isFalse();
+ }
+ }
+ }
+
@Test
void schemaShouldEnforceSingletonSettingsKeyTest() throws Exception {
try (Connection connection = DriverManager.getConnection(
