This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new 8a09821cc5 Add TCK tests for replay bounds across a deletion gap
(#3076)
8a09821cc5 is described below
commit 8a09821cc545ae241a86a4a10319d1c4df93ea34
Author: Diego D. Mucciolo <[email protected]>
AuthorDate: Tue Jun 16 00:11:17 2026 -0300
Add TCK tests for replay bounds across a deletion gap (#3076)
Motivation:
pekko-persistence-jdbc #517 added regression tests for replaying messages
whose range overlaps a deleted prefix. Port the same coverage to the shared
persistence TCK so all journal implementations meet one contract.
---
.../pekko/persistence/journal/JournalSpec.scala | 50 ++++++++++++++++++++++
1 file changed, 50 insertions(+)
diff --git
a/persistence-tck/src/main/scala/org/apache/pekko/persistence/journal/JournalSpec.scala
b/persistence-tck/src/main/scala/org/apache/pekko/persistence/journal/JournalSpec.scala
index 5388c35351..64b683204d 100644
---
a/persistence-tck/src/main/scala/org/apache/pekko/persistence/journal/JournalSpec.scala
+++
b/persistence-tck/src/main/scala/org/apache/pekko/persistence/journal/JournalSpec.scala
@@ -247,6 +247,56 @@ abstract class JournalSpec(config: Config)
journal ! ReplayMessages(0, Long.MaxValue, Long.MaxValue, pid,
receiverProbe.ref)
receiverProbe.expectMsg(RecoverySuccess(highestSequenceNr = 5L))
}
+
+ "replay all surviving messages when the lower bound equals the last
deleted sequence number" in {
+ val deleteProbe = TestProbe()
+ journal ! DeleteMessagesTo(pid, 3L, deleteProbe.ref)
+ deleteProbe.expectMsg(DeleteMessagesSuccess(3L))
+
+ journal ! ReplayMessages(3L, Long.MaxValue, Long.MaxValue, pid,
receiverProbe.ref)
+ receiverProbe.expectMsg(replayedMessage(4))
+ receiverProbe.expectMsg(replayedMessage(5))
+ receiverProbe.expectMsg(RecoverySuccess(highestSequenceNr = 5L))
+ }
+
+ "replay surviving messages within bounds when the replay window spans a
deleted prefix" in {
+ val deleteProbe = TestProbe()
+ journal ! DeleteMessagesTo(pid, 3L, deleteProbe.ref)
+ deleteProbe.expectMsg(DeleteMessagesSuccess(3L))
+
+ journal ! ReplayMessages(1, 4, Long.MaxValue, pid, receiverProbe.ref)
+ receiverProbe.expectMsg(replayedMessage(4))
+ receiverProbe.expectMsg(RecoverySuccess(highestSequenceNr = 5L))
+ }
+
+ "return only recovery success when the upper bound falls within a deleted
prefix" in {
+ val deleteProbe = TestProbe()
+ journal ! DeleteMessagesTo(pid, 3L, deleteProbe.ref)
+ deleteProbe.expectMsg(DeleteMessagesSuccess(3L))
+
+ journal ! ReplayMessages(1, 2, Long.MaxValue, pid, receiverProbe.ref)
+ receiverProbe.expectMsg(RecoverySuccess(highestSequenceNr = 5L))
+ }
+
+ "return only recovery success when the upper bound equals the last deleted
sequence number" in {
+ val deleteProbe = TestProbe()
+ journal ! DeleteMessagesTo(pid, 3L, deleteProbe.ref)
+ deleteProbe.expectMsg(DeleteMessagesSuccess(3L))
+
+ journal ! ReplayMessages(1, 3, Long.MaxValue, pid, receiverProbe.ref)
+ receiverProbe.expectMsg(RecoverySuccess(highestSequenceNr = 5L))
+ }
+
+ "replay from the first surviving message when the lower bound equals the
first surviving sequence number" in {
+ val deleteProbe = TestProbe()
+ journal ! DeleteMessagesTo(pid, 3L, deleteProbe.ref)
+ deleteProbe.expectMsg(DeleteMessagesSuccess(3L))
+
+ journal ! ReplayMessages(4, Long.MaxValue, Long.MaxValue, pid,
receiverProbe.ref)
+ receiverProbe.expectMsg(replayedMessage(4))
+ receiverProbe.expectMsg(replayedMessage(5))
+ receiverProbe.expectMsg(RecoverySuccess(highestSequenceNr = 5L))
+ }
}
"A Journal optionally".may {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
