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 f274d6a0a fix: protect notification outbox deliveries from lease loss
(#2670)
f274d6a0a is described below
commit f274d6a0a39fc9229e49b3df87251e354fcc9e76
Author: xdz997 <[email protected]>
AuthorDate: Wed Sep 2 19:59:44 2026 +0800
fix: protect notification outbox deliveries from lease loss (#2670)
---
docs/studio-native-alerting-design.md | 7 +-
.../studio/cluster/metrics/AlertingProperties.java | 6 +
.../ops/alert/NotificationOutboxService.java | 211 ++++++++++++++++--
.../mapper/RmqAlertNotificationOutboxMapper.java | 5 +
server/src/main/resources/application.yml | 3 +
.../ops/alert/NotificationOutboxServiceTest.java | 244 +++++++++++++++++++++
6 files changed, 452 insertions(+), 24 deletions(-)
diff --git a/docs/studio-native-alerting-design.md
b/docs/studio-native-alerting-design.md
index 9d8d2afc2..06c0e69a6 100644
--- a/docs/studio-native-alerting-design.md
+++ b/docs/studio-native-alerting-design.md
@@ -246,12 +246,17 @@ rmq_metric_collector_lease
collector_name, owner_id, expires_at
```
-Only the active lease holder collects and evaluates. Notification outbox rows
use claimant-bound state transitions so a stale worker cannot overwrite a newer
claimant. Delivery is at-least-once: receivers should use the event and channel
identity to deduplicate a request that times out after reaching the remote
service.
+Only the active lease holder collects and evaluates. Notification outbox rows
use claimant-bound state transitions so a stale worker cannot overwrite a newer
claimant. A delivery worker renews its claim while a webhook or SMTP call is in
flight; the renewal is conditional on the claim token, and a worker that loses
the lease stops updating that row. The delivery state is committed before its
audit entry, so an audit-store failure cannot turn a completed external send
into another retry. [...]
## Notifications and Silences
The supported notification channels are DingTalk, the SMS webhook configured
in General Settings, and Email. A real event creates outbox rows for enabled
supported channels. Independent notification-channel and notification-policy
CRUD are future work.
+The dispatcher claims a row for one minute by default and renews the claim
every 20 seconds while the external
+delivery is running. Deployments with slower receivers can tune
`studio.alerting.notification-claim-timeout` and
+`studio.alerting.notification-claim-renewal-interval`; the renewal interval
must remain shorter than the claim timeout.
+`studio.alerting.notification-heartbeat-threads` bounds the daemon workers
used for these renewals.
+
Email delivery uses Spring's standard SMTP configuration. Configure
`STUDIO_ALERTING_SMTP_HOST`,
`STUDIO_ALERTING_SMTP_PORT`, `STUDIO_ALERTING_SMTP_USERNAME`,
`STUDIO_ALERTING_SMTP_PASSWORD`,
`STUDIO_ALERTING_SMTP_AUTH`, and `STUDIO_ALERTING_SMTP_STARTTLS`; recipient
addresses are configured in
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/AlertingProperties.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/AlertingProperties.java
index 763e048fa..a7773976c 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/AlertingProperties.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/AlertingProperties.java
@@ -40,4 +40,10 @@ public class AlertingProperties {
private int notificationCleanupBatchSize = 500;
/** Maximum number of cleanup batches executed during one scheduled pass.
*/
private int notificationCleanupMaxBatches = 10;
+ /** How long a notification dispatcher may hold an outbox row without
renewing it. */
+ private String notificationClaimTimeout = "PT1M";
+ /** How often an in-flight notification claim is renewed. */
+ private String notificationClaimRenewalInterval = "PT20S";
+ /** Bounded daemon threads used for notification claim renewal. */
+ private int notificationHeartbeatThreads = 2;
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NotificationOutboxService.java
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NotificationOutboxService.java
index fce7b9336..b94fe069f 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NotificationOutboxService.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/ops/alert/NotificationOutboxService.java
@@ -11,6 +11,7 @@ import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
+import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.studio.audit.OperationAuditService;
import org.apache.rocketmq.studio.cluster.metrics.AlertingProperties;
@@ -42,6 +43,12 @@ import java.util.Set;
import java.util.ArrayList;
import java.util.function.Supplier;
import java.util.UUID;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import java.net.URLEncoder;
import java.nio.charset.StandardCharsets;
import java.util.Base64;
@@ -56,7 +63,10 @@ import jakarta.mail.internet.InternetAddress;
public class NotificationOutboxService {
private static final int MAX_ATTEMPTS = 5;
private static final int BATCH_SIZE = 20;
- private static final Duration CLAIM_TIMEOUT = Duration.ofMinutes(1);
+ private static final Duration DEFAULT_CLAIM_TIMEOUT =
Duration.ofMinutes(1);
+ private static final Duration DEFAULT_CLAIM_RENEWAL_INTERVAL =
Duration.ofSeconds(20);
+ private static final int DEFAULT_HEARTBEAT_THREADS = 2;
+ private static final int MAX_HEARTBEAT_THREADS = 4;
private static final ObjectMapper JSON = new ObjectMapper();
private final RmqAlertNotificationOutboxMapper mapper;
@@ -67,12 +77,15 @@ public class NotificationOutboxService {
private final RestTemplate restTemplate;
private final Supplier<JavaMailSender> mailSender;
private final AlertingProperties alertingProperties;
+ private final Duration claimTimeout;
+ private final Duration claimRenewalInterval;
+ private final ScheduledExecutorService heartbeatExecutor;
NotificationOutboxService(RmqAlertNotificationOutboxMapper mapper,
SettingsRepository settingsRepository,
AlertSilenceService silenceService, AlertRepository
alertRepository,
OperationAuditService operationAuditService) {
this(mapper, settingsRepository, silenceService, alertRepository,
operationAuditService, newClient(),
- () -> null, new AlertingProperties());
+ () -> null, new AlertingProperties(), null);
}
@Autowired
@@ -81,21 +94,21 @@ public class NotificationOutboxService {
OperationAuditService operationAuditService,
ObjectProvider<JavaMailSender> mailSender,
AlertingProperties alertingProperties) {
this(mapper, settingsRepository, silenceService, alertRepository,
operationAuditService, newClient(),
- mailSender::getIfAvailable, alertingProperties);
+ mailSender::getIfAvailable, alertingProperties, null);
}
NotificationOutboxService(RmqAlertNotificationOutboxMapper mapper,
SettingsRepository settingsRepository,
AlertSilenceService silenceService, AlertRepository
alertRepository,
OperationAuditService operationAuditService, RestTemplate
restTemplate) {
this(mapper, settingsRepository, silenceService, alertRepository,
operationAuditService, restTemplate,
- () -> null, new AlertingProperties());
+ () -> null, new AlertingProperties(), null);
}
NotificationOutboxService(RmqAlertNotificationOutboxMapper mapper,
SettingsRepository settingsRepository,
AlertSilenceService silenceService, AlertRepository
alertRepository,
OperationAuditService operationAuditService, AlertingProperties
alertingProperties) {
this(mapper, settingsRepository, silenceService, alertRepository,
operationAuditService, newClient(),
- () -> null, alertingProperties);
+ () -> null, alertingProperties, null);
}
NotificationOutboxService(RmqAlertNotificationOutboxMapper mapper,
SettingsRepository settingsRepository,
@@ -103,13 +116,14 @@ public class NotificationOutboxService {
OperationAuditService operationAuditService, RestTemplate
restTemplate,
Supplier<JavaMailSender> mailSender) {
this(mapper, settingsRepository, silenceService, alertRepository,
operationAuditService, restTemplate,
- mailSender, new AlertingProperties());
+ mailSender, new AlertingProperties(), null);
}
NotificationOutboxService(RmqAlertNotificationOutboxMapper mapper,
SettingsRepository settingsRepository,
AlertSilenceService silenceService, AlertRepository
alertRepository,
OperationAuditService operationAuditService, RestTemplate
restTemplate,
- Supplier<JavaMailSender> mailSender, AlertingProperties
alertingProperties) {
+ Supplier<JavaMailSender> mailSender, AlertingProperties
alertingProperties,
+ ScheduledExecutorService heartbeatExecutor) {
this.mapper = mapper;
this.settingsRepository = settingsRepository;
this.silenceService = silenceService;
@@ -118,6 +132,18 @@ public class NotificationOutboxService {
this.restTemplate = restTemplate;
this.mailSender = mailSender;
this.alertingProperties = alertingProperties == null ? new
AlertingProperties() : alertingProperties;
+ this.claimTimeout =
positiveDuration(this.alertingProperties.getNotificationClaimTimeout(),
+ DEFAULT_CLAIM_TIMEOUT, "notificationClaimTimeout");
+ Duration configuredRenewal = positiveDuration(
+ this.alertingProperties.getNotificationClaimRenewalInterval(),
+ DEFAULT_CLAIM_RENEWAL_INTERVAL,
"notificationClaimRenewalInterval");
+ this.claimRenewalInterval = safeRenewalInterval(configuredRenewal,
claimTimeout);
+ this.heartbeatExecutor = heartbeatExecutor == null ?
newHeartbeatExecutor(this.alertingProperties) : heartbeatExecutor;
+ }
+
+ @PreDestroy
+ void closeHeartbeatExecutor() {
+ heartbeatExecutor.shutdownNow();
}
public void enqueue(SystemAlertVO alert, AlertRuleVO rule) {
@@ -215,7 +241,7 @@ public class NotificationOutboxService {
throw new
org.apache.rocketmq.studio.common.exception.BusinessException(400,
"Only failed notification deliveries can be retried");
}
- recordDelivery(row, "RETRY_ALERT_NOTIFICATION_MANUALLY", "SUCCESS",
null);
+ recordDeliverySafely(row, "RETRY_ALERT_NOTIFICATION_MANUALLY",
"SUCCESS", null);
}
public NotificationDeliveryBulkRetryResult
retryFailedDeliveries(List<Long> deliveryIds) {
@@ -260,15 +286,20 @@ public class NotificationOutboxService {
@Scheduled(fixedDelayString =
"${studio.alerting.notification-dispatch-interval:PT10S}")
public void dispatch() {
- LocalDateTime now = utcNow();
- LocalDateTime staleBefore = now.minus(CLAIM_TIMEOUT);
- List<RmqAlertNotificationOutbox> due = mapper.findDispatchable(now,
staleBefore, BATCH_SIZE);
+ LocalDateTime dispatchStartedAt = utcNow();
+ LocalDateTime staleBefore = dispatchStartedAt.minus(claimTimeout);
+ List<RmqAlertNotificationOutbox> due =
mapper.findDispatchable(dispatchStartedAt, staleBefore, BATCH_SIZE);
for (RmqAlertNotificationOutbox row : due) {
String claimToken = UUID.randomUUID().toString();
- if (mapper.claimForDispatch(row.getId(), now, staleBefore, now,
claimToken) != 1) {
+ // A previous delivery in this batch may have taken a long time.
Use a fresh timestamp for
+ // each claim so a row cannot be born with an already-expired
lease.
+ LocalDateTime claimStartedAt = utcNow();
+ LocalDateTime claimStaleBefore =
claimStartedAt.minus(claimTimeout);
+ if (mapper.claimForDispatch(row.getId(), claimStartedAt,
claimStaleBefore, claimStartedAt,
+ claimToken) != 1) {
continue;
}
- send(row, now, claimToken);
+ send(row, claimToken);
}
}
@@ -305,12 +336,19 @@ public class NotificationOutboxService {
return total;
}
- private void send(RmqAlertNotificationOutbox row, LocalDateTime now,
String claimToken) {
+ private void send(RmqAlertNotificationOutbox row, String claimToken) {
+ LeaseHeartbeat heartbeat = null;
+ boolean externalSendCompleted = false;
try {
+ heartbeat = new LeaseHeartbeat(row.getId(), claimToken);
+ if (!heartbeat.isLeaseOwned()) {
+ return;
+ }
+ LocalDateTime attemptStartedAt = utcNow();
SystemAlertVO alert = loadAlert(row.getAlertId());
LocalDateTime silenceEndsAt = silenceService.activeUntil(
AlertRuleVO.builder().id(alert.getRuleId()).domain(alert.getDomain()).build(),
- alert.getInstanceId(), alert.getLabels(), now);
+ alert.getInstanceId(), alert.getLabels(),
attemptStartedAt);
if (silenceEndsAt != null) {
deferUntilSilenceEnds(row, silenceEndsAt, claimToken);
return;
@@ -321,22 +359,42 @@ public class NotificationOutboxService {
} else {
sendWebhook(settings, alert, row.getChannel(), message(row,
alert));
}
+ externalSendCompleted = true;
+ if (!heartbeat.isLeaseOwned()) {
+ log.warn("Notification delivery {} lost its claim after the
external send completed", row.getId());
+ return;
+ }
+ LocalDateTime deliveredAt = utcNow();
if (!updateClaimed(row, claimToken, new
UpdateWrapper<RmqAlertNotificationOutbox>()
- .set("status",
NotificationOutboxStatus.DELIVERED.name()).set("delivered_at", now)
+ .set("status",
NotificationOutboxStatus.DELIVERED.name()).set("delivered_at", deliveredAt)
.set("sending_started_at", null)
- .set("last_error", null))) {
+ .set("last_error", null).set("claim_token", null))) {
return;
}
- recordDelivery(row, "DELIVER_ALERT_NOTIFICATION", "SUCCESS", null);
+ recordDeliverySafely(row, "DELIVER_ALERT_NOTIFICATION", "SUCCESS",
null);
} catch (Exception error) {
- retry(row, now, claimToken, error.getMessage());
+ if (externalSendCompleted) {
+ // The receiver has already accepted the notification. A
transient database failure while
+ // recording that outcome must leave the row for stale-claim
recovery, not trigger another send.
+ log.warn("Notification delivery {} completed externally but
its state update failed: {}", row.getId(),
+ error.getMessage());
+ } else if (heartbeat == null || heartbeat.isLeaseOwned()) {
+ retry(row, utcNow(), claimToken, error.getMessage());
+ } else {
+ log.warn("Notification delivery {} failed after its claim was
lost: {}", row.getId(),
+ error.getMessage());
+ }
+ } finally {
+ if (heartbeat != null) {
+ heartbeat.close();
+ }
}
}
private void deferUntilSilenceEnds(RmqAlertNotificationOutbox row,
LocalDateTime silenceEndsAt, String claimToken) {
updateClaimed(row, claimToken, new
UpdateWrapper<RmqAlertNotificationOutbox>()
.set("status",
NotificationOutboxStatus.PENDING.name()).set("next_attempt_at", silenceEndsAt)
- .set("sending_started_at", null));
+ .set("sending_started_at", null).set("claim_token", null));
}
private void sendWebhook(GeneralSettingsVO settings, SystemAlertVO alert,
String channel, String content) {
@@ -455,10 +513,10 @@ public class NotificationOutboxService {
: NotificationOutboxStatus.RETRY_WAIT).name())
.set("next_attempt_at", now.plusSeconds(Math.min(300, 5L <<
Math.min(attempts - 1, 5))))
.set("sending_started_at", null)
- .set("last_error", abbreviate(error)))) {
+ .set("last_error", abbreviate(error)).set("claim_token",
null))) {
return;
}
- recordDelivery(row, exhausted ? "FAIL_ALERT_NOTIFICATION" :
"RETRY_ALERT_NOTIFICATION",
+ recordDeliverySafely(row, exhausted ? "FAIL_ALERT_NOTIFICATION" :
"RETRY_ALERT_NOTIFICATION",
exhausted ? "FAILURE" : "RETRYING", abbreviate(error));
log.warn("Alert notification {} for event {}: {}", exhausted ?
"failed" : "will retry", row.getAlertId(), error);
}
@@ -468,6 +526,15 @@ public class NotificationOutboxService {
"alertId=" + row.getAlertId() + ", channel=" +
row.getChannel(), result, error);
}
+ private void recordDeliverySafely(RmqAlertNotificationOutbox row, String
operation, String result, String error) {
+ try {
+ recordDelivery(row, operation, result, error);
+ } catch (RuntimeException auditFailure) {
+ log.warn("Failed to record notification delivery audit for {}:
{}", row.getId(),
+ auditFailure.getMessage());
+ }
+ }
+
private static String abbreviate(String value) {
if (value == null) return "Delivery failed";
return value.length() > 1000 ? value.substring(0, 1000) : value;
@@ -475,7 +542,105 @@ public class NotificationOutboxService {
private boolean updateClaimed(RmqAlertNotificationOutbox row, String
claimToken,
UpdateWrapper<RmqAlertNotificationOutbox> updates) {
- return mapper.update(null, updates.eq("id",
row.getId()).eq("claim_token", claimToken)) == 1;
+ return mapper.update(null, updates.eq("id", row.getId()).eq("status",
NotificationOutboxStatus.SENDING.name())
+ .eq("claim_token", claimToken)) == 1;
+ }
+
+ private static Duration positiveDuration(String configured, Duration
fallback, String propertyName) {
+ if (configured == null || configured.isBlank()) {
+ return fallback;
+ }
+ try {
+ Duration parsed = Duration.parse(configured.trim());
+ if (!parsed.isZero() && !parsed.isNegative()) {
+ return parsed;
+ }
+ } catch (RuntimeException ignored) {
+ // Fall through to the safe default below.
+ }
+ log.warn("Invalid {} {}; using {}", propertyName, configured,
fallback);
+ return fallback;
+ }
+
+ private static Duration safeRenewalInterval(Duration configured, Duration
timeout) {
+ if (configured.compareTo(timeout) < 0) {
+ return configured;
+ }
+ Duration fallback = timeout.dividedBy(2);
+ if (fallback.isZero() || fallback.isNegative()) {
+ fallback = Duration.ofMillis(1);
+ }
+ log.warn("Notification claim renewal interval {} must be shorter than
claim timeout {}; using {}",
+ configured, timeout, fallback);
+ return fallback;
+ }
+
+ private static ScheduledExecutorService
newHeartbeatExecutor(AlertingProperties properties) {
+ int configuredThreads = properties == null ? DEFAULT_HEARTBEAT_THREADS
+ : properties.getNotificationHeartbeatThreads();
+ int threads = Math.max(1, Math.min(MAX_HEARTBEAT_THREADS,
configuredThreads));
+ ThreadFactory factory = runnable -> {
+ Thread thread = new Thread(runnable,
"studio-notification-lease-heartbeat");
+ thread.setDaemon(true);
+ return thread;
+ };
+ return Executors.newScheduledThreadPool(threads, factory);
+ }
+
+ private final class LeaseHeartbeat implements AutoCloseable {
+ private final Long rowId;
+ private final String claimToken;
+ private final AtomicBoolean leaseOwned = new AtomicBoolean(true);
+ private final AtomicBoolean closed = new AtomicBoolean(false);
+ private final ScheduledFuture<?> future;
+
+ private LeaseHeartbeat(Long rowId, String claimToken) {
+ this.rowId = rowId;
+ this.claimToken = claimToken;
+ this.future = schedule();
+ }
+
+ private ScheduledFuture<?> schedule() {
+ try {
+ long intervalMillis = Math.max(1,
claimRenewalInterval.toMillis());
+ return heartbeatExecutor.scheduleAtFixedRate(this::renew,
intervalMillis, intervalMillis,
+ TimeUnit.MILLISECONDS);
+ } catch (RuntimeException schedulingFailure) {
+ loseLease("unable to schedule renewal: " +
schedulingFailure.getMessage());
+ return null;
+ }
+ }
+
+ private void renew() {
+ if (closed.get() || !leaseOwned.get()) {
+ return;
+ }
+ LocalDateTime renewedAt = utcNow();
+ try {
+ if (mapper.renewClaim(rowId, claimToken, renewedAt) != 1) {
+ loseLease("the database no longer recognizes this claim");
+ }
+ } catch (RuntimeException renewalFailure) {
+ loseLease("renewal failed: " + renewalFailure.getMessage());
+ }
+ }
+
+ private void loseLease(String reason) {
+ if (leaseOwned.compareAndSet(true, false)) {
+ log.warn("Notification delivery {} lost its claim: {}", rowId,
reason);
+ }
+ }
+
+ private boolean isLeaseOwned() {
+ return leaseOwned.get();
+ }
+
+ @Override
+ public void close() {
+ if (closed.compareAndSet(false, true) && future != null) {
+ future.cancel(false);
+ }
+ }
}
private static LocalDateTime utcNow() {
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/persistence/mapper/RmqAlertNotificationOutboxMapper.java
b/server/src/main/java/org/apache/rocketmq/studio/persistence/mapper/RmqAlertNotificationOutboxMapper.java
index 33a1af646..0642a6d0c 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/persistence/mapper/RmqAlertNotificationOutboxMapper.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/persistence/mapper/RmqAlertNotificationOutboxMapper.java
@@ -45,6 +45,11 @@ public interface RmqAlertNotificationOutboxMapper extends
BaseMapper<RmqAlertNot
@Param("staleBefore") LocalDateTime staleBefore,
@Param("claimedAt") LocalDateTime claimedAt,
@Param("claimToken") String claimToken);
+ @Update("UPDATE rmq_alert_notification_outbox SET sending_started_at =
#{renewedAt} "
+ + "WHERE id = #{id} AND status = 'SENDING' AND claim_token =
#{claimToken}")
+ int renewClaim(@Param("id") Long id, @Param("claimToken") String
claimToken,
+ @Param("renewedAt") LocalDateTime renewedAt);
+
@Select("<script>"
+ "SELECT o.id, o.alert_id AS alertId, o.channel, o.status,
o.attempt_count AS attemptCount, "
+ "o.next_attempt_at AS nextAttemptAt, o.last_error AS lastError,
o.delivered_at AS deliveredAt, "
diff --git a/server/src/main/resources/application.yml
b/server/src/main/resources/application.yml
index 235837078..9a586acbb 100644
--- a/server/src/main/resources/application.yml
+++ b/server/src/main/resources/application.yml
@@ -92,6 +92,9 @@ studio:
notification-cleanup-interval:
${STUDIO_ALERTING_NOTIFICATION_CLEANUP_INTERVAL:PT1H}
notification-cleanup-batch-size:
${STUDIO_ALERTING_NOTIFICATION_CLEANUP_BATCH_SIZE:500}
notification-cleanup-max-batches:
${STUDIO_ALERTING_NOTIFICATION_CLEANUP_MAX_BATCHES:10}
+ notification-claim-timeout:
${STUDIO_ALERTING_NOTIFICATION_CLAIM_TIMEOUT:PT1M}
+ notification-claim-renewal-interval:
${STUDIO_ALERTING_NOTIFICATION_CLAIM_RENEWAL_INTERVAL:PT20S}
+ notification-heartbeat-threads:
${STUDIO_ALERTING_NOTIFICATION_HEARTBEAT_THREADS:2}
llm:
token: ${RMQ_LLM_TOKEN:}
anthropic-base-url: ${RMQ_ANTHROPIC_BASE_URL:}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NotificationOutboxServiceTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NotificationOutboxServiceTest.java
index e99a5da25..541d5b2d2 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NotificationOutboxServiceTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NotificationOutboxServiceTest.java
@@ -30,12 +30,23 @@ import java.util.Arrays;
import java.util.List;
import java.util.Locale;
import java.util.Optional;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.argThat;
+import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import static org.assertj.core.api.Assertions.assertThat;
@@ -518,4 +529,237 @@ class NotificationOutboxServiceTest {
assertThat(result.getSucceededIds()).containsExactly(8L);
assertThat(result.getFailures()).containsKey(9L);
}
+
+ @Test
+ void renewsClaimWhileEmailDeliveryIsStillInFlightTest() throws Exception {
+ RmqAlertNotificationOutboxMapper mapper =
mock(RmqAlertNotificationOutboxMapper.class);
+ SettingsRepository settings = mock(SettingsRepository.class);
+ AlertRepository alerts = mock(AlertRepository.class);
+ JavaMailSender mailSender = mock(JavaMailSender.class);
+ ScheduledExecutorService heartbeatExecutor =
mock(ScheduledExecutorService.class);
+ ScheduledFuture<?> heartbeatFuture = mock(ScheduledFuture.class);
+ AtomicReference<Runnable> renewal = new AtomicReference<>();
+ when(heartbeatExecutor.scheduleAtFixedRate(any(Runnable.class),
anyLong(), anyLong(), any(TimeUnit.class)))
+ .thenAnswer(invocation -> {
+ renewal.set(invocation.getArgument(0));
+ return heartbeatFuture;
+ });
+
+ RmqAlertNotificationOutbox row = new RmqAlertNotificationOutbox();
+ row.setId(8L);
+ row.setAlertId(9L);
+ row.setChannel("email");
+ row.setStatus("PENDING");
+ row.setAttemptCount(0);
+ when(mapper.findDispatchable(any(LocalDateTime.class),
any(LocalDateTime.class), any(Integer.class)))
+ .thenReturn(List.of(row));
+ when(mapper.claimForDispatch(any(), any(LocalDateTime.class),
any(LocalDateTime.class),
+ any(LocalDateTime.class), anyString())).thenReturn(1);
+ when(mapper.renewClaim(eq(8L), anyString(),
any(LocalDateTime.class))).thenReturn(1);
+ when(mapper.update(any(), any())).thenReturn(1);
+
when(alerts.findAlertById(9L)).thenReturn(Optional.of(SystemAlertVO.builder().id(9L)
+
.level(AlertLevel.warning).title("Lag").description("high").instanceId("local").build()));
+
when(settings.loadGeneralSettings()).thenReturn(GeneralSettingsVO.builder()
+ .emailRecipients("[email protected]").build());
+
+ CountDownLatch sendStarted = new CountDownLatch(1);
+ CountDownLatch releaseSend = new CountDownLatch(1);
+ org.mockito.Mockito.doAnswer(invocation -> {
+ sendStarted.countDown();
+ if (!releaseSend.await(5, TimeUnit.SECONDS)) {
+ throw new IllegalStateException("test send timed out");
+ }
+ return null;
+ }).when(mailSender).send(any(SimpleMailMessage.class));
+
+ AlertingProperties properties = new AlertingProperties();
+ properties.setNotificationClaimTimeout("PT1M");
+ properties.setNotificationClaimRenewalInterval("PT10S");
+ NotificationOutboxService service = new
NotificationOutboxService(mapper, settings,
+ mock(AlertSilenceService.class), alerts,
mock(OperationAuditService.class), new RestTemplate(),
+ () -> mailSender, properties, heartbeatExecutor);
+
+ ExecutorService dispatcher = Executors.newSingleThreadExecutor();
+ try {
+ Future<?> dispatch = dispatcher.submit(service::dispatch);
+ org.junit.jupiter.api.Assertions.assertTrue(sendStarted.await(5,
TimeUnit.SECONDS));
+ org.junit.jupiter.api.Assertions.assertNotNull(renewal.get());
+ renewal.get().run();
+ releaseSend.countDown();
+ dispatch.get(5, TimeUnit.SECONDS);
+ } finally {
+ releaseSend.countDown();
+ dispatcher.shutdownNow();
+ service.closeHeartbeatExecutor();
+ }
+
+ verify(mapper).renewClaim(eq(8L), anyString(),
any(LocalDateTime.class));
+ verify(mapper).update(any(), any());
+ verify(heartbeatFuture).cancel(false);
+ }
+
+ @Test
+ void doesNotRetryAfterDeliveryWhenAuditRecordingFailsTest() {
+ RmqAlertNotificationOutboxMapper mapper =
mock(RmqAlertNotificationOutboxMapper.class);
+ SettingsRepository settings = mock(SettingsRepository.class);
+ AlertRepository alerts = mock(AlertRepository.class);
+ OperationAuditService audit = mock(OperationAuditService.class);
+ ScheduledExecutorService heartbeatExecutor =
mock(ScheduledExecutorService.class);
+ ScheduledFuture<?> heartbeatFuture = mock(ScheduledFuture.class);
+ when(heartbeatExecutor.scheduleAtFixedRate(any(Runnable.class),
anyLong(), anyLong(), any(TimeUnit.class)))
+ .thenAnswer(invocation -> heartbeatFuture);
+ org.mockito.Mockito.doAnswer(invocation -> {
+ if
("DELIVER_ALERT_NOTIFICATION".equals(invocation.getArgument(0))) {
+ throw new IllegalStateException("audit database unavailable");
+ }
+ return null;
+ }).when(audit).record(anyString(), anyString(), anyString(), any(),
anyString(), anyString(), any());
+
+ RmqAlertNotificationOutbox row = new RmqAlertNotificationOutbox();
+ row.setId(8L);
+ row.setAlertId(9L);
+ row.setChannel("dingtalk");
+ row.setStatus("PENDING");
+ row.setAttemptCount(0);
+ when(mapper.findDispatchable(any(LocalDateTime.class),
any(LocalDateTime.class), any(Integer.class)))
+ .thenReturn(List.of(row));
+ when(mapper.claimForDispatch(any(), any(LocalDateTime.class),
any(LocalDateTime.class),
+ any(LocalDateTime.class), anyString())).thenReturn(1);
+ when(mapper.update(any(), any())).thenReturn(1);
+
when(alerts.findAlertById(9L)).thenReturn(Optional.of(SystemAlertVO.builder().id(9L)
+
.level(AlertLevel.warning).title("Lag").description("high").instanceId("local").build()));
+
when(settings.loadGeneralSettings()).thenReturn(GeneralSettingsVO.builder()
+ .dingtalkWebhook("https://example.com/hook";).build());
+
+ RestTemplate client = new RestTemplate();
+ MockRestServiceServer server =
MockRestServiceServer.bindTo(client).build();
+ server.expect(once(), requestTo("https://example.com/hook";))
+ .andRespond(withSuccess("{\"errcode\":0}",
MediaType.APPLICATION_JSON));
+
+ NotificationOutboxService service = new
NotificationOutboxService(mapper, settings,
+ mock(AlertSilenceService.class), alerts, audit, client, () ->
null,
+ new AlertingProperties(), heartbeatExecutor);
+ try {
+ service.dispatch();
+ } finally {
+ service.closeHeartbeatExecutor();
+ }
+
+ server.verify();
+ verify(mapper, times(1)).update(any(), any());
+ verify(audit).record("DELIVER_ALERT_NOTIFICATION",
"ALERT_NOTIFICATION", "8", null,
+ "alertId=9, channel=dingtalk", "SUCCESS", null);
+ }
+
+ @Test
+ void abandonsStateUpdateWhenDeliveryLeaseIsLostDuringSendTest() throws
Exception {
+ RmqAlertNotificationOutboxMapper mapper =
mock(RmqAlertNotificationOutboxMapper.class);
+ SettingsRepository settings = mock(SettingsRepository.class);
+ AlertRepository alerts = mock(AlertRepository.class);
+ OperationAuditService audit = mock(OperationAuditService.class);
+ ScheduledExecutorService heartbeatExecutor =
mock(ScheduledExecutorService.class);
+ ScheduledFuture<?> heartbeatFuture = mock(ScheduledFuture.class);
+ AtomicReference<Runnable> renewal = new AtomicReference<>();
+ when(heartbeatExecutor.scheduleAtFixedRate(any(Runnable.class),
anyLong(), anyLong(), any(TimeUnit.class)))
+ .thenAnswer(invocation -> {
+ renewal.set(invocation.getArgument(0));
+ return heartbeatFuture;
+ });
+
+ RmqAlertNotificationOutbox row = new RmqAlertNotificationOutbox();
+ row.setId(8L);
+ row.setAlertId(9L);
+ row.setChannel("email");
+ row.setStatus("PENDING");
+ row.setAttemptCount(0);
+ when(mapper.findDispatchable(any(LocalDateTime.class),
any(LocalDateTime.class), any(Integer.class)))
+ .thenReturn(List.of(row));
+ when(mapper.claimForDispatch(any(), any(LocalDateTime.class),
any(LocalDateTime.class),
+ any(LocalDateTime.class), anyString())).thenReturn(1);
+ when(mapper.renewClaim(eq(8L), anyString(),
any(LocalDateTime.class))).thenReturn(0);
+
when(alerts.findAlertById(9L)).thenReturn(Optional.of(SystemAlertVO.builder().id(9L)
+
.level(AlertLevel.warning).title("Lag").description("high").instanceId("local").build()));
+
when(settings.loadGeneralSettings()).thenReturn(GeneralSettingsVO.builder()
+ .emailRecipients("[email protected]").build());
+
+ CountDownLatch sendStarted = new CountDownLatch(1);
+ CountDownLatch releaseSend = new CountDownLatch(1);
+ JavaMailSender mailSender = mock(JavaMailSender.class);
+ org.mockito.Mockito.doAnswer(invocation -> {
+ sendStarted.countDown();
+ if (!releaseSend.await(5, TimeUnit.SECONDS)) {
+ throw new IllegalStateException("test send timed out");
+ }
+ return null;
+ }).when(mailSender).send(any(SimpleMailMessage.class));
+
+ NotificationOutboxService service = new
NotificationOutboxService(mapper, settings,
+ mock(AlertSilenceService.class), alerts, audit, new
RestTemplate(), () -> mailSender,
+ new AlertingProperties(), heartbeatExecutor);
+ ExecutorService dispatcher = Executors.newSingleThreadExecutor();
+ try {
+ Future<?> dispatch = dispatcher.submit(service::dispatch);
+ org.junit.jupiter.api.Assertions.assertTrue(sendStarted.await(5,
TimeUnit.SECONDS));
+ org.junit.jupiter.api.Assertions.assertNotNull(renewal.get());
+ renewal.get().run();
+ releaseSend.countDown();
+ dispatch.get(5, TimeUnit.SECONDS);
+ } finally {
+ releaseSend.countDown();
+ dispatcher.shutdownNow();
+ service.closeHeartbeatExecutor();
+ }
+
+ verify(mapper).renewClaim(eq(8L), anyString(),
any(LocalDateTime.class));
+ verify(mapper, never()).update(any(), any());
+ verify(audit, never()).record(any(), any(), any(), any(), any(),
any(), any());
+ verify(heartbeatFuture).cancel(false);
+ }
+
+ @Test
+ void doesNotRetryWhenDeliveryStateWriteFailsAfterExternalSuccessTest() {
+ RmqAlertNotificationOutboxMapper mapper =
mock(RmqAlertNotificationOutboxMapper.class);
+ SettingsRepository settings = mock(SettingsRepository.class);
+ AlertRepository alerts = mock(AlertRepository.class);
+ OperationAuditService audit = mock(OperationAuditService.class);
+ ScheduledExecutorService heartbeatExecutor =
mock(ScheduledExecutorService.class);
+ ScheduledFuture<?> heartbeatFuture = mock(ScheduledFuture.class);
+ when(heartbeatExecutor.scheduleAtFixedRate(any(Runnable.class),
anyLong(), anyLong(), any(TimeUnit.class)))
+ .thenAnswer(invocation -> heartbeatFuture);
+
+ RmqAlertNotificationOutbox row = new RmqAlertNotificationOutbox();
+ row.setId(8L);
+ row.setAlertId(9L);
+ row.setChannel("dingtalk");
+ row.setStatus("PENDING");
+ row.setAttemptCount(0);
+ when(mapper.findDispatchable(any(LocalDateTime.class),
any(LocalDateTime.class), any(Integer.class)))
+ .thenReturn(List.of(row));
+ when(mapper.claimForDispatch(any(), any(LocalDateTime.class),
any(LocalDateTime.class),
+ any(LocalDateTime.class), anyString())).thenReturn(1);
+ when(mapper.update(any(), any())).thenThrow(new
IllegalStateException("outbox database unavailable"));
+
when(alerts.findAlertById(9L)).thenReturn(Optional.of(SystemAlertVO.builder().id(9L)
+
.level(AlertLevel.warning).title("Lag").description("high").instanceId("local").build()));
+
when(settings.loadGeneralSettings()).thenReturn(GeneralSettingsVO.builder()
+ .dingtalkWebhook("https://example.com/hook";).build());
+
+ RestTemplate client = new RestTemplate();
+ MockRestServiceServer server =
MockRestServiceServer.bindTo(client).build();
+ server.expect(once(), requestTo("https://example.com/hook";))
+ .andRespond(withSuccess("{\"errcode\":0}",
MediaType.APPLICATION_JSON));
+
+ NotificationOutboxService service = new
NotificationOutboxService(mapper, settings,
+ mock(AlertSilenceService.class), alerts, audit, client, () ->
null,
+ new AlertingProperties(), heartbeatExecutor);
+ try {
+ service.dispatch();
+ } finally {
+ service.closeHeartbeatExecutor();
+ }
+
+ server.verify();
+ verify(mapper).update(any(), any());
+ verify(audit, never()).record(any(), any(), any(), any(), any(),
any(), any());
+ verify(heartbeatFuture).cancel(false);
+ }
}
