Skip to content

Commit 7d6526f

Browse files
J0nasMayrJ0nasMayr
committed
add test for concurrency and schedule functionality
1 parent 6fa769d commit 7d6526f

1 file changed

Lines changed: 235 additions & 0 deletions

File tree

‎src/test/java/it/aboutbits/springboot/emailservice/lib/application/SendScheduledEmailsTest.java‎

Lines changed: 235 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,235 @@
1+
package it.aboutbits.springboot.emailservice.lib.application;
2+
3+
import it.aboutbits.springboot.emailservice.lib.EmailState;
4+
import it.aboutbits.springboot.emailservice.lib.jpa.EmailRepository;
5+
import it.aboutbits.springboot.emailservice.support.database.WithPostgres;
6+
import it.aboutbits.springboot.emailservice.support.database.factory.EmailFactory;
7+
import jakarta.mail.internet.MimeMessage;
8+
import org.jspecify.annotations.NullMarked;
9+
import org.junit.jupiter.api.BeforeEach;
10+
import org.junit.jupiter.api.Test;
11+
import org.springframework.beans.factory.annotation.Autowired;
12+
import org.springframework.boot.test.context.SpringBootTest;
13+
import org.springframework.mail.MailSendException;
14+
import org.springframework.mail.javamail.JavaMailSender;
15+
import org.springframework.test.context.bean.override.mockito.MockitoSpyBean;
16+
17+
import java.time.OffsetDateTime;
18+
import java.util.concurrent.CountDownLatch;
19+
import java.util.concurrent.Executors;
20+
import java.util.concurrent.TimeUnit;
21+
import java.util.stream.IntStream;
22+
23+
import static org.assertj.core.api.Assertions.assertThat;
24+
import static org.mockito.ArgumentMatchers.any;
25+
import static org.mockito.Mockito.doNothing;
26+
import static org.mockito.Mockito.doThrow;
27+
import static org.mockito.Mockito.times;
28+
import static org.mockito.Mockito.verify;
29+
30+
@SpringBootTest(properties = {
31+
"aboutbits.emailservice.scheduling.batch-size=5",
32+
"aboutbits.emailservice.scheduling.max-attempts=3",
33+
"aboutbits.emailservice.scheduling.stuck-sending-recovery-threshold=PT5M",
34+
"aboutbits.emailservice.scheduling.interval=30000"
35+
})
36+
@WithPostgres
37+
@NullMarked
38+
class SendScheduledEmailsTest {
39+
private static final int BATCH_SIZE = 5;
40+
private static final int MAX_ATTEMPTS = 3;
41+
private static final int SCHEDULER_INTERVAL_SECONDS = 30;
42+
43+
@MockitoSpyBean
44+
JavaMailSender javaMailSender;
45+
46+
@Autowired
47+
EmailRepository emailRepository;
48+
49+
@Autowired
50+
SendScheduledEmails sendScheduledEmails;
51+
52+
@BeforeEach
53+
void setup() {
54+
doNothing().when(javaMailSender).send(any(MimeMessage.class));
55+
}
56+
57+
@Test
58+
void givenPendingEmails_sendEmails_shouldMarkAllAsSent() {
59+
emailRepository.saveAll(IntStream.range(0, 3).mapToObj(_ -> EmailFactory.once().build()).toList());
60+
61+
sendScheduledEmails.sendEmails();
62+
63+
assertThat(emailRepository.findAll())
64+
.hasSize(3)
65+
.allMatch(email -> email.getState() == EmailState.SENT)
66+
.allMatch(email -> email.getExecutionStartTime() != null)
67+
.allMatch(email -> email.getExecutionEndTime() != null)
68+
.allMatch(email -> email.getAttempts() == 1);
69+
verify(javaMailSender, times(3)).send(any(MimeMessage.class));
70+
}
71+
72+
@Test
73+
void givenMoreEmailsThanBatchSize_sendEmails_shouldProcessOnlyBatchSizePerPass() {
74+
var pending = 3 * BATCH_SIZE;
75+
emailRepository.saveAll(IntStream.range(0, pending).mapToObj(_ -> EmailFactory.once().build()).toList());
76+
77+
sendScheduledEmails.sendEmails();
78+
79+
assertThat(emailRepository.findAll())
80+
.filteredOn(email -> email.getState() == EmailState.SENT)
81+
.hasSize(BATCH_SIZE);
82+
verify(javaMailSender, times(BATCH_SIZE)).send(any(MimeMessage.class));
83+
}
84+
85+
@Test
86+
void givenRetryableFailure_sendEmails_shouldRescheduleWithBackoffAndIncrementAttempts() {
87+
doThrow(new MailSendException("smtp blip")).when(javaMailSender).send(any(MimeMessage.class));
88+
emailRepository.save(EmailFactory.once().build());
89+
90+
var beforePass = OffsetDateTime.now();
91+
sendScheduledEmails.sendEmails();
92+
93+
assertThat(emailRepository.findAll())
94+
.singleElement()
95+
.satisfies(email -> {
96+
assertThat(email.getState()).isEqualTo(EmailState.PENDING);
97+
assertThat(email.getAttempts()).isEqualTo(1);
98+
// First retry backoff: attempts (1) * 2 * scheduler interval.
99+
assertThat(email.getScheduledAt())
100+
.isAfterOrEqualTo(beforePass.plusSeconds(2L * SCHEDULER_INTERVAL_SECONDS));
101+
assertThat(email.getErrorMessage()).contains("smtp blip");
102+
assertThat(email.getExecutionEndTime()).isNotNull();
103+
});
104+
}
105+
106+
@Test
107+
void givenAttemptsAlreadyAtBudget_sendEmails_shouldEscalateToError() {
108+
doThrow(new MailSendException("smtp down")).when(javaMailSender).send(any(MimeMessage.class));
109+
emailRepository.save(
110+
EmailFactory.once()
111+
.attempts(MAX_ATTEMPTS)
112+
.scheduledAt(OffsetDateTime.now().minusSeconds(30))
113+
.build()
114+
);
115+
116+
sendScheduledEmails.sendEmails();
117+
118+
assertThat(emailRepository.findAll())
119+
.singleElement()
120+
.satisfies(email -> {
121+
// The atomic claim UPDATE incremented attempts from MAX_ATTEMPTS to MAX_ATTEMPTS+1,
122+
// which is > threshold, so the failure escalates to ERROR.
123+
assertThat(email.getState()).isEqualTo(EmailState.ERROR);
124+
assertThat(email.getAttempts()).isEqualTo(MAX_ATTEMPTS + 1);
125+
assertThat(email.getErrorMessage()).contains("smtp down");
126+
});
127+
}
128+
129+
@Test
130+
void givenErrorRow_sendEmails_shouldNotRetry() {
131+
emailRepository.save(
132+
EmailFactory.once()
133+
.state(EmailState.ERROR)
134+
.attempts(MAX_ATTEMPTS + 1)
135+
.scheduledAt(OffsetDateTime.now().minusMinutes(1))
136+
.errorMessage("previous permanent failure")
137+
.build()
138+
);
139+
140+
sendScheduledEmails.sendEmails();
141+
142+
assertThat(emailRepository.findAll())
143+
.singleElement()
144+
.satisfies(email -> {
145+
assertThat(email.getState()).isEqualTo(EmailState.ERROR);
146+
assertThat(email.getAttempts()).isEqualTo(MAX_ATTEMPTS + 1);
147+
});
148+
verify(javaMailSender, times(0)).send(any(MimeMessage.class));
149+
}
150+
151+
@Test
152+
void givenStuckSendingRow_sendEmails_shouldRecoverAndResend() {
153+
var staleStart = OffsetDateTime.now().minusMinutes(10);
154+
emailRepository.save(
155+
EmailFactory.once()
156+
.state(EmailState.SENDING)
157+
.attempts(1)
158+
.executionStartTime(staleStart)
159+
.scheduledAt(OffsetDateTime.now().minusSeconds(60))
160+
.build()
161+
);
162+
163+
sendScheduledEmails.sendEmails();
164+
165+
assertThat(emailRepository.findAll())
166+
.singleElement()
167+
.satisfies(email -> {
168+
assertThat(email.getState()).isEqualTo(EmailState.SENT);
169+
assertThat(email.getAttempts()).isEqualTo(2);
170+
assertThat(email.getExecutionStartTime()).isAfter(staleStart);
171+
});
172+
verify(javaMailSender, times(1)).send(any(MimeMessage.class));
173+
}
174+
175+
@Test
176+
void givenFreshSendingRowWithinThreshold_sendEmails_shouldNotStealFromOtherPod() {
177+
// executionStartTime within the 5-minute threshold means another pod is legitimately
178+
// sending this right now; we must not re-claim it.
179+
var recentStart = OffsetDateTime.now().minusSeconds(30);
180+
emailRepository.save(
181+
EmailFactory.once()
182+
.state(EmailState.SENDING)
183+
.attempts(1)
184+
.executionStartTime(recentStart)
185+
.scheduledAt(OffsetDateTime.now().minusSeconds(60))
186+
.build()
187+
);
188+
189+
sendScheduledEmails.sendEmails();
190+
191+
assertThat(emailRepository.findAll())
192+
.singleElement()
193+
.satisfies(email -> {
194+
assertThat(email.getState()).isEqualTo(EmailState.SENDING);
195+
assertThat(email.getAttempts()).isEqualTo(1);
196+
assertThat(email.getExecutionStartTime()).isEqualTo(recentStart);
197+
});
198+
verify(javaMailSender, times(0)).send(any(MimeMessage.class));
199+
}
200+
201+
@Test
202+
void givenConcurrentPasses_sendEmails_shouldSendEachEmailExactlyOnce() throws Exception {
203+
var totalEmails = 25;
204+
var workerThreads = 4;
205+
var passesPerWorker = 10;
206+
emailRepository.saveAll(IntStream.range(0, totalEmails).mapToObj(_ -> EmailFactory.once().build()).toList());
207+
208+
var startGate = new CountDownLatch(1);
209+
var executor = Executors.newFixedThreadPool(workerThreads);
210+
try {
211+
for (var t = 0; t < workerThreads; t++) {
212+
executor.submit(() -> {
213+
startGate.await();
214+
for (var i = 0; i < passesPerWorker; i++) {
215+
sendScheduledEmails.sendEmails();
216+
}
217+
return null;
218+
});
219+
}
220+
startGate.countDown();
221+
executor.shutdown();
222+
assertThat(executor.awaitTermination(60, TimeUnit.SECONDS)).isTrue();
223+
} finally {
224+
if (!executor.isTerminated()) {
225+
executor.shutdownNow();
226+
}
227+
}
228+
229+
assertThat(emailRepository.findAll())
230+
.hasSize(totalEmails)
231+
.allMatch(email -> email.getState() == EmailState.SENT)
232+
.allMatch(email -> email.getAttempts() == 1);
233+
verify(javaMailSender, times(totalEmails)).send(any(MimeMessage.class));
234+
}
235+
}

0 commit comments

Comments
 (0)