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 b7704c6db fix(alert): bound SMTP sockets and stamp outbox rows in UTC
(#6121)
b7704c6db is described below
commit b7704c6dbfed0f59c3a052be65c685c3c75da2af
Author: wizcraft_kris <[email protected]>
AuthorDate: Mon Oct 12 16:14:13 2026 +0800
fix(alert): bound SMTP sockets and stamp outbox rows in UTC (#6121)
Two fixes to the notification outbox, whose email path had no time budget
and whose bookkeeping timestamps were written by a clock the service does not
control.
- #6072 JavaMail's three socket timeouts default to infinite, and
`NotificationOutboxService.sendEmail` (`:477-489`) calls `sender.send(message)`
directly from the `@Scheduled` dispatcher at `:306` and the cleanup job at
`:325`; the repo configures no `spring.task.scheduling` pool and defines no
custom `TaskScheduler` bean, so Boot's default pool size of 1 means an SMTP
relay that accepts the connection and never sends a 220 banner pins the
backend's only scheduling thread forever — s [...]
- #6121 `db/schema.sql:402-403` lets the database session clock fill
`gmt_create` and `gmt_modified` (`DEFAULT CURRENT_TIMESTAMP` / `ON UPDATE
CURRENT_TIMESTAMP`) while the retention cutoff is a JVM UTC instant
(`NotificationOutboxService.java:340`, consumed by
`RmqAlertNotificationOutboxMapper.java:27-29`), and nothing forces the two
clocks to agree — `application.yml:11` defaults to `serverTimezone=UTC` while
`deploy/docker-compose.yml:86` ships `Asia/Shanghai` — so terminal deliver [...]
Tests: a new `SmtpTimeoutTest` pins the shipped 5000/10000/10000 values,
proves all three properties are deployment-overridable, and reproduces the
wedge against a real loopback `ServerSocket` that accepts and then stays
silent, asserting a `MailSendException` with a `SocketTimeoutException` root
cause from a send running on a daemon thread that the test wakes by closing the
accepted socket; `NotificationOutboxServiceTest` adds four cases and
`NotificationOutboxMapperIntegrationTest` [...]
Folded: #6072 was closed as folded into this commit (same author).
---
docs/studio-native-alerting-design.md | 8 +
.../ops/alert/NotificationOutboxService.java | 14 +-
.../entity/RmqAlertNotificationOutbox.java | 1 +
.../mapper/RmqAlertNotificationOutboxMapper.java | 7 +-
server/src/main/resources/application.yml | 3 +
.../NotificationOutboxMapperIntegrationTest.java | 28 +++
.../ops/alert/NotificationOutboxServiceTest.java | 147 +++++++++++++++
.../rocketmq/studio/ops/alert/SmtpTimeoutTest.java | 198 +++++++++++++++++++++
8 files changed, 402 insertions(+), 4 deletions(-)
diff --git a/docs/studio-native-alerting-design.md
b/docs/studio-native-alerting-design.md
index 7b1f75d53..e3242fa8d 100644
--- a/docs/studio-native-alerting-design.md
+++ b/docs/studio-native-alerting-design.md
@@ -268,6 +268,14 @@ Email delivery uses Spring's standard SMTP configuration.
Configure `STUDIO_ALER
`STUDIO_ALERTING_SMTP_AUTH`, and `STUDIO_ALERTING_SMTP_STARTTLS`; recipient
addresses are configured in
General Settings. Missing SMTP configuration leaves the outbox row retryable
and records the delivery failure.
+Socket timeouts are bounded by default and tunable through
`STUDIO_ALERTING_SMTP_CONNECTION_TIMEOUT` (5000 ms),
+`STUDIO_ALERTING_SMTP_READ_TIMEOUT` (10000 ms) and
`STUDIO_ALERTING_SMTP_WRITE_TIMEOUT` (10000 ms). They matter
+because `@Scheduled` work runs on the single scheduler thread Spring Boot uses
by default (`spring.task.scheduling.pool.size`
+is not raised in `application.yml`), and that thread also drives metrics
collection, authentication session cleanup,
+query-history pruning, conversation cleanup and outbox retention. An
unreachable SMTP peer that never resets the
+connection would otherwise hold it for as long as the OS socket timeout
allows, freezing every other scheduled task
+with it. A timed-out send leaves the outbox row retryable, which is the same
path a missing configuration takes.
+
```text
PENDING -> SENDING -> DELIVERED
-> RETRY_WAIT -> FAILED
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 955eb3178..7e16786ff 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
@@ -197,7 +197,12 @@ public class NotificationOutboxService {
row.setStatus(NotificationOutboxStatus.PENDING.name());
row.setAttemptCount(0);
row.setMessageContent(AlertNotificationTemplate.render(rule.getNotificationTemplate(),
alert, rule));
- row.setNextAttemptAt(silenceEndsAt == null ? utcNow() :
silenceEndsAt);
+ LocalDateTime now = utcNow();
+ row.setNextAttemptAt(silenceEndsAt == null ? now : silenceEndsAt);
+ // The retention cutoff is computed in UTC; leaving these to the
database defaults would
+ // stamp them with the session clock (Asia/Shanghai in the compose
deploy) instead.
+ row.setGmtCreate(now);
+ row.setGmtModified(now);
mapper.insert(row);
}
}
@@ -251,7 +256,8 @@ public class NotificationOutboxService {
int updated = mapper.update(null, new
UpdateWrapper<RmqAlertNotificationOutbox>()
.set("status",
NotificationOutboxStatus.PENDING.name()).set("attempt_count", 0)
.set("next_attempt_at", now).set("sending_started_at",
null).set("claim_token", null)
- .set("last_error", null).eq("id", deliveryId).eq("status",
NotificationOutboxStatus.FAILED.name()));
+ .set("last_error", null).set("gmt_modified", now)
+ .eq("id", deliveryId).eq("status",
NotificationOutboxStatus.FAILED.name()));
if (updated != 1) {
throw new
org.apache.rocketmq.studio.common.exception.BusinessException(400,
"Only failed notification deliveries can be retried");
@@ -567,6 +573,10 @@ public class NotificationOutboxService {
private boolean updateClaimed(RmqAlertNotificationOutbox row, String
claimToken,
UpdateWrapper<RmqAlertNotificationOutbox> updates) {
+ // Every claimed state change stamps gmt_modified from the same UTC
clock as the retention
+ // cutoff. Without this the ON UPDATE CURRENT_TIMESTAMP default
rewrites it in the database
+ // session's timezone and the purge compares two clocks.
+ updates.set("gmt_modified", utcNow());
return mapper.update(null, updates.eq("id", row.getId()).eq("status",
NotificationOutboxStatus.SENDING.name())
.eq("claim_token", claimToken)) == 1;
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqAlertNotificationOutbox.java
b/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqAlertNotificationOutbox.java
index 3484bc62f..5d984819f 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqAlertNotificationOutbox.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/persistence/entity/RmqAlertNotificationOutbox.java
@@ -27,5 +27,6 @@ public class RmqAlertNotificationOutbox {
private String lastError;
private String messageContent;
private LocalDateTime deliveredAt;
+ private LocalDateTime gmtCreate;
private LocalDateTime gmtModified;
}
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 6243da06c..c59e0a24d 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
@@ -38,14 +38,17 @@ public interface RmqAlertNotificationOutboxMapper extends
BaseMapper<RmqAlertNot
@Param("staleBefore") LocalDateTime staleBefore, @Param("limit")
int limit);
@Update("UPDATE rmq_alert_notification_outbox SET status = 'SENDING',
sending_started_at = #{claimedAt}, "
- + "claim_token = #{claimToken} "
+ + "claim_token = #{claimToken}, gmt_modified = #{claimedAt} "
+ "WHERE id = #{id} AND ((status IN ('PENDING', 'RETRY_WAIT') AND
next_attempt_at <= #{now}) "
+ "OR (status = 'SENDING' AND (sending_started_at IS NULL OR
sending_started_at <= #{staleBefore})))")
int claimForDispatch(@Param("id") Long id, @Param("now") LocalDateTime now,
@Param("staleBefore") LocalDateTime staleBefore,
@Param("claimedAt") LocalDateTime claimedAt,
@Param("claimToken") String claimToken);
- @Update("UPDATE rmq_alert_notification_outbox SET sending_started_at =
#{renewedAt} "
+ // gmt_modified is set explicitly so the caller's clock wins over the
column's
+ // ON UPDATE CURRENT_TIMESTAMP default (the database session's clock).
+ @Update("UPDATE rmq_alert_notification_outbox SET sending_started_at =
#{renewedAt}, "
+ + "gmt_modified = #{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);
diff --git a/server/src/main/resources/application.yml
b/server/src/main/resources/application.yml
index cf2f41b8c..a6d174bef 100644
--- a/server/src/main/resources/application.yml
+++ b/server/src/main/resources/application.yml
@@ -22,6 +22,9 @@ spring:
properties:
mail.smtp.auth: true
mail.smtp.starttls.enable: true
+ mail.smtp.connectiontimeout:
${STUDIO_ALERTING_SMTP_CONNECTION_TIMEOUT:5000}
+ mail.smtp.timeout: ${STUDIO_ALERTING_SMTP_READ_TIMEOUT:10000}
+ mail.smtp.writetimeout: ${STUDIO_ALERTING_SMTP_WRITE_TIMEOUT:10000}
ai:
mcp:
server:
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NotificationOutboxMapperIntegrationTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NotificationOutboxMapperIntegrationTest.java
index 776ff92bf..b96f8fecc 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NotificationOutboxMapperIntegrationTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NotificationOutboxMapperIntegrationTest.java
@@ -139,6 +139,34 @@ class NotificationOutboxMapperIntegrationTest {
}
}
+ @Test
+ void claimAndRenewStampGmtModifiedFromTheCallerClockTest() {
+ List<Long> alertIds = List.of(ALERT_ID_BASE + 201);
+ cleanup(alertIds);
+ try {
+ insert(alertIds.get(0), "dingtalk",
NotificationOutboxStatus.PENDING, null, LocalDateTime.now());
+ Long id = mapper.selectList(new
QueryWrapper<RmqAlertNotificationOutbox>()
+ .in("alert_id", alertIds)).get(0).getId();
+ // Whole-second values: the datetime column drops sub-second
precision on read-back.
+ LocalDateTime claimedAt = LocalDateTime.of(2026, 9, 30, 8, 0, 0);
+ LocalDateTime renewedAt = LocalDateTime.of(2026, 9, 30, 8, 0, 20);
+
+ // The claims carry the dispatch clock. Without an explicit
gmt_modified the column falls
+ // back to ON UPDATE CURRENT_TIMESTAMP - the database session
clock, not the caller's.
+ // The due window looks a minute ahead: datetime rounds the insert
timestamp up to the
+ // whole second, so a wall-clock "now" can fall a fraction before
the stored value.
+ LocalDateTime claimNow = LocalDateTime.now().plusMinutes(1);
+ assertThat(mapper.claimForDispatch(id, claimNow,
claimNow.minusMinutes(5),
+ claimedAt, "token-a")).isEqualTo(1);
+
assertThat(mapper.selectById(id).getGmtModified()).isEqualTo(claimedAt);
+
+ assertThat(mapper.renewClaim(id, "token-a",
renewedAt)).isEqualTo(1);
+
assertThat(mapper.selectById(id).getGmtModified()).isEqualTo(renewedAt);
+ } finally {
+ cleanup(alertIds);
+ }
+ }
+
@Test
void deleteTerminalBeforeShouldRespectBatchLimitTest() {
LocalDateTime expired = LocalDateTime.now().minusDays(31);
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 aea9578c0..2e9ca586d 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
@@ -901,4 +901,151 @@ class NotificationOutboxServiceTest {
server.verify();
}
+
+ @Test
+ void enqueuedRowsStampCreateAndModifyWithUtcTest() {
+ TimeZone previous = TimeZone.getDefault();
+ try {
+ TimeZone.setDefault(TimeZone.getTimeZone("Asia/Shanghai"));
+ RmqAlertNotificationOutboxMapper mapper =
mock(RmqAlertNotificationOutboxMapper.class);
+ AlertSilenceService silences = mock(AlertSilenceService.class);
+ AlertRuleVO rule =
AlertRuleVO.builder().id(4L).domain(AlertDomain.BUSINESS)
+ .channels(List.of("dingtalk")).build();
+ SystemAlertVO alert =
SystemAlertVO.builder().id(9L).level(AlertLevel.warning).title("Lag")
+ .description("lag is
high").instanceId("local").time(LocalDateTime.now()).build();
+ when(silences.activeUntil(rule, "local", java.util.Map.of(),
alert.getTime())).thenReturn(null);
+
+ new NotificationOutboxService(mapper,
mock(SettingsRepository.class), silences,
+ mock(AlertRepository.class),
mock(OperationAuditService.class)).enqueue(alert, rule);
+
+ org.mockito.ArgumentCaptor<RmqAlertNotificationOutbox> row =
+
org.mockito.ArgumentCaptor.forClass(RmqAlertNotificationOutbox.class);
+ verify(mapper).insert(row.capture());
+ // Without an explicit stamp the database fills these from DEFAULT
CURRENT_TIMESTAMP -
+ // the session clock, Asia/Shanghai in the compose deploy - while
the retention cutoff
+ // is computed in UTC, so the two sides of the comparison live in
different timezones.
+ assertUtcRecent(row.getValue().getGmtCreate());
+ assertUtcRecent(row.getValue().getGmtModified());
+ } finally {
+ TimeZone.setDefault(previous);
+ }
+ }
+
+ @Test
+ void deliveredTerminalWriteStampsGmtModifiedInUtcTest() {
+ TimeZone previous = TimeZone.getDefault();
+ try {
+ TimeZone.setDefault(TimeZone.getTimeZone("Asia/Shanghai"));
+ RmqAlertNotificationOutboxMapper mapper =
mock(RmqAlertNotificationOutboxMapper.class);
+ SettingsRepository settings = mock(SettingsRepository.class);
+ AlertRepository alerts = mock(AlertRepository.class);
+ RestTemplate client = new RestTemplate();
+ MockRestServiceServer server =
MockRestServiceServer.bindTo(client).build();
+ 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")
+ .labels(java.util.Map.of()).build()));
+
when(settings.loadGeneralSettings()).thenReturn(GeneralSettingsVO.builder()
+ .dingtalkWebhook("https://example.com/hook";).build());
+ server.expect(once(), requestTo("https://example.com/hook";))
+ .andRespond(withSuccess("{\"errcode\":0}",
MediaType.APPLICATION_JSON));
+
+ new NotificationOutboxService(mapper, settings,
mock(AlertSilenceService.class), alerts,
+ mock(OperationAuditService.class), client).dispatch();
+
+ server.verify();
+ assertUtcRecent(stampedAt(capturedUpdate(mapper), "gmt_modified"));
+ } finally {
+ TimeZone.setDefault(previous);
+ }
+ }
+
+ @Test
+ void exhaustedDeliveryStampsGmtModifiedOnTheFailedWriteTest() {
+ TimeZone previous = TimeZone.getDefault();
+ try {
+ TimeZone.setDefault(TimeZone.getTimeZone("Asia/Shanghai"));
+ RmqAlertNotificationOutboxMapper mapper =
mock(RmqAlertNotificationOutboxMapper.class);
+ // The fifth attempt ends the delivery as FAILED, the row the
retention purge qualifies
+ // by gmt_modified - so this write is the one that decides when
the row is deleted.
+ RmqAlertNotificationOutbox row = new RmqAlertNotificationOutbox();
+ row.setId(8L);
+ row.setAlertId(9L);
+ row.setChannel("dingtalk");
+ row.setStatus("PENDING");
+ row.setAttemptCount(4);
+ 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);
+ AlertRepository alerts = mock(AlertRepository.class);
+
when(alerts.findAlertById(9L)).thenReturn(Optional.of(SystemAlertVO.builder().id(9L).build()));
+
+ new NotificationOutboxService(mapper,
mock(SettingsRepository.class), mock(AlertSilenceService.class),
+ alerts, mock(OperationAuditService.class)).dispatch();
+
+ assertUtcRecent(stampedAt(capturedUpdate(mapper), "gmt_modified"));
+ } finally {
+ TimeZone.setDefault(previous);
+ }
+ }
+
+ @Test
+ void manualRetryStampsGmtModifiedInUtcTest() {
+ TimeZone previous = TimeZone.getDefault();
+ try {
+ TimeZone.setDefault(TimeZone.getTimeZone("Asia/Shanghai"));
+ RmqAlertNotificationOutboxMapper mapper =
mock(RmqAlertNotificationOutboxMapper.class);
+ RmqAlertNotificationOutbox row = new RmqAlertNotificationOutbox();
+ row.setId(8L);
+ row.setAlertId(9L);
+ row.setChannel("dingtalk");
+ row.setStatus(NotificationOutboxStatus.FAILED.name());
+ when(mapper.selectById(8L)).thenReturn(row);
+ when(mapper.update(org.mockito.ArgumentMatchers.isNull(),
any(UpdateWrapper.class))).thenReturn(1);
+
+ new NotificationOutboxService(mapper,
mock(SettingsRepository.class), mock(AlertSilenceService.class),
+ mock(AlertRepository.class),
mock(OperationAuditService.class)).retryFailedDelivery(8L);
+
+ assertUtcRecent(stampedAt(capturedUpdate(mapper), "gmt_modified"));
+ } finally {
+ TimeZone.setDefault(previous);
+ }
+ }
+
+ @SuppressWarnings({"unchecked", "rawtypes"})
+ private static UpdateWrapper<RmqAlertNotificationOutbox>
capturedUpdate(RmqAlertNotificationOutboxMapper mapper) {
+ org.mockito.ArgumentCaptor<UpdateWrapper> captor =
org.mockito.ArgumentCaptor.forClass(UpdateWrapper.class);
+ verify(mapper).update(any(), captor.capture());
+ return captor.getValue();
+ }
+
+ private static LocalDateTime
stampedAt(UpdateWrapper<RmqAlertNotificationOutbox> wrapper, String column) {
+ String sqlSet = wrapper.getSqlSet() == null ? "" : wrapper.getSqlSet();
+ assertThat(sqlSet).as("UPDATE set clause for " +
column).contains(column + "=");
+ java.util.regex.Matcher placeholder = java.util.regex.Pattern
+ .compile(column +
"=#\\{ew\\.paramNameValuePairs\\.([^}]+)\\}").matcher(sqlSet);
+ if (!placeholder.find()) {
+ return null;
+ }
+ Object value =
wrapper.getParamNameValuePairs().get(placeholder.group(1));
+ return value instanceof LocalDateTime ? (LocalDateTime) value : null;
+ }
+
+ private static void assertUtcRecent(LocalDateTime stamp) {
+ assertThat(stamp).as("explicit UTC timestamp").isNotNull();
+ assertThat(Duration.between(stamp.toInstant(ZoneOffset.UTC),
java.time.Instant.now()).abs())
+ .isLessThan(Duration.ofSeconds(2));
+ }
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/SmtpTimeoutTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/SmtpTimeoutTest.java
new file mode 100644
index 000000000..ce19cb4ee
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/SmtpTimeoutTest.java
@@ -0,0 +1,198 @@
+/*
+ * 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.io.IOException;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.net.SocketTimeoutException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.autoconfigure.AutoConfigurations;
+import org.springframework.boot.mail.autoconfigure.MailSenderAutoConfiguration;
+import
org.springframework.boot.test.context.ConfigDataApplicationContextInitializer;
+import org.springframework.boot.test.context.runner.ApplicationContextRunner;
+import org.springframework.mail.MailSendException;
+import org.springframework.mail.SimpleMailMessage;
+import org.springframework.mail.javamail.JavaMailSender;
+import org.springframework.mail.javamail.JavaMailSenderImpl;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.fail;
+
+/**
+ * The alert e-mail path hands {@code sender.send(...)} to the scheduler
thread that every
+ * {@code @Scheduled} job shares, so the JavaMail socket timeouts configured in
+ * {@code application.yml} are what keep a slow or wedged SMTP relay from
stalling the whole
+ * backend. These tests pin the shipped defaults, the deployment overrides,
and the behaviour of
+ * a send aimed at a peer that accepts the connection and never answers.
+ */
+class SmtpTimeoutTest {
+
+ private static final int SEND_GUARD_SECONDS = 25;
+
+ private final ApplicationContextRunner runner = new
ApplicationContextRunner()
+ .withInitializer(new ConfigDataApplicationContextInitializer())
+
.withConfiguration(AutoConfigurations.of(MailSenderAutoConfiguration.class));
+
+ @Test
+ void shippedMailPropertiesBoundEverySocketTimeoutTest() {
+ runner.run(context -> {
+ Properties properties =
context.getBean(JavaMailSenderImpl.class).getJavaMailProperties();
+
+ assertThat(properties)
+ .as("spring.mail.properties must bound every JavaMail
socket timeout")
+ .containsKeys("mail.smtp.connectiontimeout",
"mail.smtp.timeout", "mail.smtp.writetimeout");
+ assertThat(timeoutMillis(properties,
"mail.smtp.connectiontimeout")).isEqualTo(5000);
+ assertThat(timeoutMillis(properties,
"mail.smtp.timeout")).isEqualTo(10000);
+ assertThat(timeoutMillis(properties,
"mail.smtp.writetimeout")).isEqualTo(10000);
+ });
+ }
+
+ @Test
+ void socketTimeoutsAreDeploymentOverridableTest() {
+ runner.withPropertyValues(
+ "STUDIO_ALERTING_SMTP_CONNECTION_TIMEOUT=1234",
+ "STUDIO_ALERTING_SMTP_READ_TIMEOUT=2345",
+ "STUDIO_ALERTING_SMTP_WRITE_TIMEOUT=3456")
+ .run(context -> {
+ Properties properties =
context.getBean(JavaMailSenderImpl.class).getJavaMailProperties();
+
+ assertThat(timeoutMillis(properties,
"mail.smtp.connectiontimeout")).isEqualTo(1234);
+ assertThat(timeoutMillis(properties,
"mail.smtp.timeout")).isEqualTo(2345);
+ assertThat(timeoutMillis(properties,
"mail.smtp.writetimeout")).isEqualTo(3456);
+ });
+ }
+
+ @Test
+ void sendFailsFastWhenSmtpServerNeverAnswersTest() throws Exception {
+ try (SilentSmtpServer server = new SilentSmtpServer()) {
+ // A deployment that actually sends alerts has credentials
configured, so the client
+ // opens the transport and waits for the SMTP greeting instead of
failing before any I/O.
+ runner.withPropertyValues("spring.mail.host=127.0.0.1",
"spring.mail.port=" + server.port(),
+ "spring.mail.username=probe-user",
"spring.mail.password=probe-secret")
+ .run(context -> {
+ JavaMailSender sender =
context.getBean(JavaMailSender.class);
+ SimpleMailMessage message = new SimpleMailMessage();
+ message.setFrom("[email protected]");
+ message.setTo("[email protected]");
+ message.setSubject("[RocketMQ Studio] timeout probe");
+ message.setText("probe");
+
+ assertSendTerminates(sender, message);
+ });
+ }
+ }
+
+ /**
+ * Runs the send on a daemon thread so a wedged socket cannot outlive the
test JVM, and fails
+ * with the defect statement when the call is still blocked once the guard
expires.
+ */
+ private void assertSendTerminates(JavaMailSender sender, SimpleMailMessage
message) {
+ ExecutorService executor = Executors.newSingleThreadExecutor(worker ->
{
+ Thread thread = new Thread(worker, "smtp-send-under-test");
+ thread.setDaemon(true);
+ return thread;
+ });
+ try {
+ Future<?> send = executor.submit(() -> sender.send(message));
+ try {
+ send.get(SEND_GUARD_SECONDS, TimeUnit.SECONDS);
+ } catch (TimeoutException blocked) {
+ fail("sender.send() was still blocked after " +
SEND_GUARD_SECONDS
+ + "s against an SMTP peer that accepted the connection
and never replied: "
+ + "spring.mail.properties sets no
mail.smtp.connectiontimeout / mail.smtp.timeout / "
+ + "mail.smtp.writetimeout, so a wedged relay pins its
caller forever (in production "
+ + "that caller is the single @Scheduled dispatcher
shared by all scheduled jobs)");
+ } catch (ExecutionException rejected) {
+ assertThat(rejected.getCause())
+ .as("the send must fail with a bounded mail error
rather than hang")
+ .isInstanceOf(MailSendException.class)
+ .hasRootCauseInstanceOf(SocketTimeoutException.class);
+ } catch (InterruptedException interrupted) {
+ Thread.currentThread().interrupt();
+ fail("interrupted while waiting for the SMTP send to return");
+ }
+ } finally {
+ executor.shutdownNow();
+ }
+ }
+
+ private static int timeoutMillis(Properties properties, String key) {
+ Object value = properties.get(key);
+ assertThat(value).as("JavaMail property %s must be set",
key).isNotNull();
+ return Integer.parseInt(String.valueOf(value));
+ }
+
+ /**
+ * Accepts SMTP connections and then stays silent: no 220 banner, no
reply. JavaMail connects
+ * successfully and waits for the greeting, which is exactly the read that
an unbounded
+ * {@code mail.smtp.timeout} never finishes.
+ */
+ private static final class SilentSmtpServer implements AutoCloseable {
+
+ private final ServerSocket server;
+ private final List<Socket> accepted = new ArrayList<>();
+ private final Thread acceptor;
+
+ private SilentSmtpServer() throws IOException {
+ this.server = new ServerSocket(0, 1,
InetAddress.getLoopbackAddress());
+ this.acceptor = new Thread(this::acceptQuietly,
"silent-smtp-acceptor");
+ this.acceptor.setDaemon(true);
+ this.acceptor.start();
+ }
+
+ private void acceptQuietly() {
+ while (!server.isClosed()) {
+ try {
+ accepted.add(server.accept());
+ } catch (IOException closed) {
+ return;
+ }
+ }
+ }
+
+ private int port() {
+ return server.getLocalPort();
+ }
+
+ @Override
+ public void close() {
+ try {
+ server.close();
+ } catch (IOException ignored) {
+ // best effort: only used to unblock a hung client at the end
of the test
+ }
+ for (Socket socket : accepted) {
+ try {
+ socket.close();
+ } catch (IOException ignored) {
+ // best effort: closing the accepted sockets wakes the
client read
+ }
+ }
+ }
+ }
+}
