This is an automated email from the ASF dual-hosted git repository.
pjfanning 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 bfe2a5789e feat: support TimestampOffset in
PersistenceTestKitReadJournal tag and slice queries (#3615)
bfe2a5789e is described below
commit bfe2a5789ecf16183d752c2c083bc8a960ff7def
Author: PJ Fanning <[email protected]>
AuthorDate: Sun Oct 11 09:14:58 2026 +0100
feat: support TimestampOffset in PersistenceTestKitReadJournal tag and
slice queries (#3615)
* feat: support TimestampOffset in PersistenceTestKitReadJournal tag and
slice queries
eventsByTag, currentEventsByTag, eventsBySlices and currentEventsBySlices
used to throw UnsupportedOperationException for anything but NoOffset.
They now accept a TimestampOffset and resume after it. Other offset types
fail with IllegalArgumentException.
Envelopes from the tag queries and the live eventsBySlices query now carry
a TimestampOffset (as currentEventsBySlices already did) instead of a
per-persistence-id Sequence, so the offset can be passed back in to resume.
Fixes #1765
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* fix: keep event timestamps in serialized test kit storage
Motivation:
Tag queries now return `TimestampOffset`, which broke tests in
persistence-typed-tests that expected `Sequence` offsets. Those tests
also showed that the offsets carried an epoch-0 timestamp:
`SerializedEventStorageImpl`, used with the default `serialize = true`,
dropped `PersistentRepr.timestamp`, so offsets could not be resumed
from correctly.
Modification:
- Store the timestamp in `SerializedEventStorageImpl.Serialized` and
restore it when reading.
- Update the tag query assertions in `EventSourcedBehaviorSpec`,
`EventSourcedEventAdapterSpec` and `EventSourcedBehaviorJavaDslTest`
to expect a `TimestampOffset`, using `inside` / `assertInstanceOf`
rather than casts.
Result:
Events read from the serializing test kit storage keep their write
timestamp, and tag query offsets point at it.
Tests:
- `persistence-typed-tests/testOnly *EventSourcedBehaviorSpec
*EventSourcedEventAdapterSpec *EventSourcedBehaviorJavaDslTest` and
`persistence-testkit/test`: passed on 2.13 and 3.3.8
- `EventSourcedBehaviorSpec` "tag events" asserts a non-zero timestamp,
which fails without the storage fix
- scalafmt and javafmt run on changed files
References:
Refs #1765
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../internal/SerializedEventStorageImpl.scala | 8 ++--
.../query/internal/EventsBySliceStage.scala | 11 +++--
.../testkit/query/internal/EventsByTagStage.scala | 11 +++--
.../testkit/query/internal/TimestampOffsets.scala | 56 ++++++++++++++++++++++
.../scaladsl/PersistenceTestKitReadJournal.scala | 46 ++++++------------
.../testkit/query/CurrentEventsBySlicesSpec.scala | 30 ++++++++++++
.../testkit/query/CurrentEventsByTagSpec.scala | 27 +++++++++++
.../testkit/query/EventsBySliceSpec.scala | 28 +++++++++++
.../testkit/query/EventsByTagSpec.scala | 25 ++++++++++
.../javadsl/EventSourcedBehaviorJavaDslTest.java | 5 +-
.../typed/scaladsl/EventSourcedBehaviorSpec.scala | 18 +++++--
.../scaladsl/EventSourcedEventAdapterSpec.scala | 8 +++-
12 files changed, 227 insertions(+), 46 deletions(-)
diff --git
a/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/internal/SerializedEventStorageImpl.scala
b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/internal/SerializedEventStorageImpl.scala
index 75b60cb326..963610d653 100644
---
a/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/internal/SerializedEventStorageImpl.scala
+++
b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/internal/SerializedEventStorageImpl.scala
@@ -33,7 +33,8 @@ private[testkit] object SerializedEventStorageImpl {
writerUuid: String,
payload: Array[Byte],
tags: Set[String],
- metadata: Option[Any])
+ metadata: Option[Any],
+ timestamp: Long)
}
/**
@@ -67,7 +68,8 @@ private[testkit] class SerializedEventStorageImpl(system:
ActorSystem) extends E
writerUuid = pr.writerUuid,
payload = s.toBinary(payload),
tags = tags,
- metadata = pr.metadata)
+ metadata = pr.metadata,
+ timestamp = pr.timestamp)
}
/**
@@ -83,7 +85,7 @@ private[testkit] class SerializedEventStorageImpl(system:
ActorSystem) extends E
sequenceNr = internal.sequenceNr,
persistenceId = internal.persistenceId,
writerUuid = internal.writerUuid,
- manifest = internal.eventAdapterManifest)
+ manifest =
internal.eventAdapterManifest).withTimestamp(internal.timestamp)
internal.metadata.fold(pr)(meta => pr.withMetadata(meta))
}
diff --git
a/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/query/internal/EventsBySliceStage.scala
b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/query/internal/EventsBySliceStage.scala
index b3c94efb78..8912be67ef 100644
---
a/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/query/internal/EventsBySliceStage.scala
+++
b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/query/internal/EventsBySliceStage.scala
@@ -21,7 +21,7 @@ import org.apache.pekko
import pekko.actor.ActorRef
import pekko.annotation.InternalApi
import pekko.persistence.Persistence
-import pekko.persistence.query.Sequence
+import pekko.persistence.query.Offset
import pekko.persistence.query.typed
import pekko.persistence.testkit.EventStorage
import pekko.persistence.testkit.PersistenceTestKitPlugin.SliceWrite
@@ -61,6 +61,7 @@ final private[pekko] class EventsBySliceStage[Event](
entityType: String,
minSlice: Int,
maxSlice: Int,
+ offset: Offset,
storage: EventStorage,
persistence: Persistence
) extends GraphStage[SourceShape[typed.EventEnvelope[Event]]] {
@@ -73,6 +74,7 @@ final private[pekko] class EventsBySliceStage[Event](
new GraphStageLogicWithLogging(shape) with OutHandler {
private var state = Option.empty[State]
private var stageActorRef: ActorRef = null
+ private val isAfterOffset = TimestampOffsets.isAfter(offset)
override def preStart(): Unit = {
stageActorRef = getStageActor(receiveNotifications).ref
materializer.system.eventStream.subscribe(stageActorRef,
classOf[SliceWrite])
@@ -98,7 +100,10 @@ final private[pekko] class EventsBySliceStage[Event](
val maybeNextEvent = storage.tryRead(entityType, repr =>
shouldFilter(repr.persistenceId))
.sortBy(pr => (pr.timestamp, pr.sequenceNr))
.find { pr =>
- state.forall(_.isAfter(pr.timestamp, pr.sequenceNr))
+ state match {
+ case Some(s) => s.isAfter(pr.timestamp, pr.sequenceNr)
+ case None => isAfterOffset(pr)
+ }
}
log.debug("tryPush available. State {} event {}", state,
maybeNextEvent)
@@ -106,7 +111,7 @@ final private[pekko] class EventsBySliceStage[Event](
maybeNextEvent.foreach { pr =>
val slice = persistence.sliceForPersistenceId(pr.persistenceId)
push(out,
- new typed.EventEnvelope[Event](Sequence(pr.sequenceNr),
pr.persistenceId, pr.sequenceNr,
+ new typed.EventEnvelope[Event](TimestampOffsets.offsetFor(pr),
pr.persistenceId, pr.sequenceNr,
Some(pr.payload.asInstanceOf[Event]), pr.timestamp,
pr.metadata, entityType, slice))
state = Some(State(pr.timestamp, pr.sequenceNr))
diff --git
a/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/query/internal/EventsByTagStage.scala
b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/query/internal/EventsByTagStage.scala
index 301e26433f..5a67f2b5f7 100644
---
a/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/query/internal/EventsByTagStage.scala
+++
b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/query/internal/EventsByTagStage.scala
@@ -22,7 +22,7 @@ import pekko.actor.ActorRef
import pekko.annotation.InternalApi
import pekko.persistence.journal.Tagged
import pekko.persistence.query.EventEnvelope
-import pekko.persistence.query.Sequence
+import pekko.persistence.query.Offset
import pekko.persistence.testkit.EventStorage
import pekko.persistence.testkit.PersistenceTestKitPlugin
import pekko.persistence.testkit.PersistenceTestKitPlugin.TagWrite
@@ -59,6 +59,7 @@ private[pekko] object EventsByTagStage {
@InternalApi
final private[pekko] class EventsByTagStage(
tag: String,
+ offset: Offset,
storage: EventStorage)
extends GraphStage[SourceShape[EventEnvelope]] {
import EventsByTagStage._
@@ -70,6 +71,7 @@ final private[pekko] class EventsByTagStage(
new GraphStageLogicWithLogging(shape) with OutHandler {
private var state = Option.empty[State]
private var stageActorRef: ActorRef = null
+ private val isAfterOffset = TimestampOffsets.isAfter(offset)
override def preStart(): Unit = {
stageActorRef = getStageActor(receiveNotifications).ref
materializer.system.eventStream.subscribe(stageActorRef,
classOf[PersistenceTestKitPlugin.TagWrite])
@@ -91,14 +93,17 @@ final private[pekko] class EventsByTagStage(
.tryReadByTag(tag)
.sortBy(pr => (pr.timestamp, pr.sequenceNr))
.find { pr =>
- state.forall(_.isAfter(pr.timestamp, pr.sequenceNr))
+ state match {
+ case Some(s) => s.isAfter(pr.timestamp, pr.sequenceNr)
+ case None => isAfterOffset(pr)
+ }
}
log.debug("tryPush available. State {} event {}", state,
maybeNextEvent)
maybeNextEvent.foreach { pr =>
push(out,
- EventEnvelope(Sequence(pr.sequenceNr), pr.persistenceId,
pr.sequenceNr,
+ EventEnvelope(TimestampOffsets.offsetFor(pr), pr.persistenceId,
pr.sequenceNr,
pr.payload match {
case Tagged(payload, _) => payload
case payload => payload
diff --git
a/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/query/internal/TimestampOffsets.scala
b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/query/internal/TimestampOffsets.scala
new file mode 100644
index 0000000000..b39cbb783a
--- /dev/null
+++
b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/query/internal/TimestampOffsets.scala
@@ -0,0 +1,56 @@
+/*
+ * 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.pekko.persistence.testkit.query.internal
+
+import java.time.Instant
+import java.time.temporal.ChronoUnit
+
+import org.apache.pekko
+import pekko.annotation.InternalApi
+import pekko.persistence.PersistentRepr
+import pekko.persistence.query.Offset
+import pekko.persistence.query.TimestampOffset
+
+/**
+ * INTERNAL API
+ *
+ * The test kit journal assigns a strictly increasing timestamp to each atomic
write,
+ * and an atomic write only contains events of a single persistence id, so
+ * `(timestamp, persistenceId, sequenceNr)` identifies the position of an
event.
+ */
+@InternalApi
+private[pekko] object TimestampOffsets {
+
+ def offsetFor(pr: PersistentRepr): TimestampOffset = {
+ val timestamp = Instant.ofEpochMilli(pr.timestamp)
+ val readTimestamp = Instant.now().truncatedTo(ChronoUnit.MICROS)
+ TimestampOffset(timestamp, readTimestamp, Map(pr.persistenceId ->
pr.sequenceNr))
+ }
+
+ /**
+ * Returns a predicate that selects the events after the given offset, which
must be
+ * a [[TimestampOffset]] or [[pekko.persistence.query.NoOffset]]. The offset
is exclusive.
+ */
+ def isAfter(offset: Offset): PersistentRepr => Boolean = {
+ val timestampOffset = TimestampOffset.toTimestampOffset(offset)
+ pr => {
+ val cmp =
Instant.ofEpochMilli(pr.timestamp).compareTo(timestampOffset.timestamp)
+ cmp > 0 || (cmp == 0 &&
timestampOffset.seen.get(pr.persistenceId).forall(pr.sequenceNr > _))
+ }
+ }
+}
diff --git
a/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/query/scaladsl/PersistenceTestKitReadJournal.scala
b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/query/scaladsl/PersistenceTestKitReadJournal.scala
index 318dd7799e..1a36848408 100644
---
a/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/query/scaladsl/PersistenceTestKitReadJournal.scala
+++
b/persistence-testkit/src/main/scala/org/apache/pekko/persistence/testkit/query/scaladsl/PersistenceTestKitReadJournal.scala
@@ -12,8 +12,6 @@
*/
package org.apache.pekko.persistence.testkit.query.scaladsl
-import java.time.Instant
-import java.time.temporal.ChronoUnit
import scala.annotation.nowarn
@@ -44,6 +42,7 @@ import
pekko.persistence.testkit.internal.InMemStorageExtension
import pekko.persistence.testkit.query.internal.EventsByPersistenceIdStage
import pekko.persistence.testkit.query.internal.EventsBySliceStage
import pekko.persistence.testkit.query.internal.EventsByTagStage
+import pekko.persistence.testkit.query.internal.TimestampOffsets
import pekko.persistence.testkit.query.internal.TypedEventsByPersistenceIdStage
import pekko.persistence.typed.PersistenceId
import pekko.stream.scaladsl.Source
@@ -90,12 +89,6 @@ final class PersistenceTestKitReadJournal(system:
ExtendedActorSystem, @nowarn("
case _ => Set.empty
}
- private def timestampOffsetFor(pr: pekko.persistence.PersistentRepr):
TimestampOffset = {
- val timestamp = Instant.ofEpochMilli(pr.timestamp)
- val readTimestamp = Instant.now().truncatedTo(ChronoUnit.MICROS)
- TimestampOffset(timestamp, readTimestamp, Map(pr.persistenceId ->
pr.sequenceNr))
- }
-
override def eventsByPersistenceId(
persistenceId: String,
fromSequenceNr: Long = 0,
@@ -134,7 +127,7 @@ final class PersistenceTestKitReadJournal(system:
ExtendedActorSystem, @nowarn("
val entityType = PersistenceId.extractEntityType(persistenceId)
Source(storage.tryRead(persistenceId, fromSequenceNr, toSequenceNr,
Long.MaxValue)).map { pr =>
typed.EventEnvelope(
- timestampOffsetFor(pr),
+ TimestampOffsets.offsetFor(pr),
persistenceId,
pr.sequenceNr,
unwrapTaggedPayload(pr.payload).asInstanceOf[Event],
@@ -148,14 +141,10 @@ final class PersistenceTestKitReadJournal(system:
ExtendedActorSystem, @nowarn("
}
override def currentEventsByTag(tag: String, offset: Offset = NoOffset):
Source[EventEnvelope, NotUsed] = {
- offset match {
- case NoOffset =>
- case _ =>
- throw new UnsupportedOperationException("Offsets not supported for
persistence test kit currentEventsByTag yet")
- }
- Source(storage.tryReadByTag(tag)).map { pr =>
+ val isAfterOffset = TimestampOffsets.isAfter(offset)
+ Source(storage.tryReadByTag(tag).filter(isAfterOffset)).map { pr =>
EventEnvelope(
- Sequence(pr.sequenceNr),
+ TimestampOffsets.offsetFor(pr),
pr.persistenceId,
pr.sequenceNr,
unwrapTaggedPayload(pr.payload),
@@ -169,21 +158,18 @@ final class PersistenceTestKitReadJournal(system:
ExtendedActorSystem, @nowarn("
minSlice: Int,
maxSlice: Int,
offset: Offset): Source[typed.EventEnvelope[Event], NotUsed] = {
- offset match {
- case NoOffset =>
- case _ =>
- throw new UnsupportedOperationException("Offsets not supported for
persistence test kit currentEventsByTag yet")
- }
+ val isAfterOffset = TimestampOffsets.isAfter(offset)
val prs = storage.tryRead(entityType,
repr => {
val pid = repr.persistenceId
val slice = persistence.sliceForPersistenceId(pid)
- PersistenceId.extractEntityType(pid) == entityType && slice >=
minSlice && slice <= maxSlice
+ PersistenceId.extractEntityType(pid) == entityType && slice >=
minSlice && slice <= maxSlice &&
+ isAfterOffset(repr)
})
Source(prs).map { pr =>
val slice = persistence.sliceForPersistenceId(pr.persistenceId)
typed.EventEnvelope(
- timestampOffsetFor(pr),
+ TimestampOffsets.offsetFor(pr),
pr.persistenceId,
pr.sequenceNr,
unwrapTaggedPayload(pr.payload).asInstanceOf[Event],
@@ -218,10 +204,9 @@ final class PersistenceTestKitReadJournal(system:
ExtendedActorSystem, @nowarn("
storage.currentPersistenceIds(afterId, limit)
override def eventsByTag(tag: String, offset: Offset = NoOffset):
Source[EventEnvelope, NotUsed] = {
- if (offset != NoOffset) {
- throw new UnsupportedOperationException("Offsets not supported for
persistence test kit currentEventsByTag yet")
- }
- Source.fromGraph(new EventsByTagStage(tag, storage))
+ // validate eagerly so that an unsupported offset type fails the call
rather than the stream
+ TimestampOffset.toTimestampOffset(offset)
+ Source.fromGraph(new EventsByTagStage(tag, offset, storage))
}
override def eventsBySlices[Event](
@@ -230,9 +215,8 @@ final class PersistenceTestKitReadJournal(system:
ExtendedActorSystem, @nowarn("
maxSlice: Int,
offset: Offset
): Source[typed.EventEnvelope[Event], NotUsed] = {
- if (offset != NoOffset) {
- throw new UnsupportedOperationException("Offsets not supported for
persistence test kit eventsBySlices yet")
- }
- Source.fromGraph(new EventsBySliceStage(entityType, minSlice, maxSlice,
storage, persistence))
+ // validate eagerly so that an unsupported offset type fails the call
rather than the stream
+ TimestampOffset.toTimestampOffset(offset)
+ Source.fromGraph(new EventsBySliceStage(entityType, minSlice, maxSlice,
offset, storage, persistence))
}
}
diff --git
a/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/query/CurrentEventsBySlicesSpec.scala
b/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/query/CurrentEventsBySlicesSpec.scala
index 82365c3b15..d737b5491b 100644
---
a/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/query/CurrentEventsBySlicesSpec.scala
+++
b/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/query/CurrentEventsBySlicesSpec.scala
@@ -21,6 +21,7 @@ import pekko.actor.typed.ActorRef
import pekko.persistence.Persistence
import pekko.persistence.query.NoOffset
import pekko.persistence.query.PersistenceQuery
+import pekko.persistence.query.Sequence
import pekko.persistence.query.TimestampOffset
import pekko.persistence.testkit.query.EventsByPersistenceIdSpec.Command
import pekko.persistence.testkit.query.EventsByPersistenceIdSpec.testBehavior
@@ -73,6 +74,35 @@ class CurrentEventsBySlicesSpec
.map(_.event) should ===(Seq("evt-1", "evt-2", "evt-3", "evt-4",
"evt-5"))
}
+ "find eventsBySlices after the given offset" in {
+ val probe = createTestProbe[Done]()
+ val ref1 = spawn(testBehavior("OffsetTest|pid-1"))
+ val ref2 = spawn(testBehavior("OffsetTest|pid-2"))
+ ref1 ! Command("evt-1", probe.ref)
+ ref1 ! Command("evt-2", probe.ref)
+ probe.receiveMessages(2)
+ ref2 ! Command("evt-3", probe.ref)
+ probe.receiveMessage()
+ ref1 ! Command("evt-4", probe.ref)
+ probe.receiveMessage()
+
+ val maxSlice = Persistence(system).numberOfSlices - 1
+ val all = queries.currentEventsBySlices[String]("OffsetTest", 0,
maxSlice, NoOffset).runWith(Sink.seq).futureValue
+ all.map(_.event) should ===(Seq("evt-1", "evt-2", "evt-3", "evt-4"))
+
+ queries
+ .currentEventsBySlices[String]("OffsetTest", 0, maxSlice,
all(1).offset)
+ .runWith(Sink.seq)
+ .futureValue
+ .map(_.event) should ===(Seq("evt-3", "evt-4"))
+ }
+
+ "reject unsupported offset types for eventsBySlices" in {
+ intercept[IllegalArgumentException] {
+ queries.currentEventsBySlices[String]("Test", 0,
Persistence(system).numberOfSlices - 1, Sequence(1L))
+ }
+ }
+
"include tags in events by slices" in {
val probe = createTestProbe[Done]()
val ref1 = spawn(testBehavior("TagTest|pid-1"))
diff --git
a/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/query/CurrentEventsByTagSpec.scala
b/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/query/CurrentEventsByTagSpec.scala
index 120d534acf..b1d620bc97 100644
---
a/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/query/CurrentEventsByTagSpec.scala
+++
b/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/query/CurrentEventsByTagSpec.scala
@@ -20,6 +20,7 @@ import
pekko.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit
import pekko.actor.typed.ActorRef
import pekko.persistence.query.NoOffset
import pekko.persistence.query.PersistenceQuery
+import pekko.persistence.query.Sequence
import pekko.persistence.testkit.query.EventsByPersistenceIdSpec.Command
import pekko.persistence.testkit.query.EventsByPersistenceIdSpec.testBehavior
import pekko.persistence.testkit.query.scaladsl.PersistenceTestKitReadJournal
@@ -62,6 +63,32 @@ class CurrentEventsByTagSpec
queries.currentEventsByTag("all",
NoOffset).runWith(Sink.seq).futureValue.map(_.event) should ===(
Seq("evt-1", "evt-2", "evt-3", "evt-4", "evt-5"))
}
+
+ "find tagged events after the given offset" in {
+ val probe = createTestProbe[Done]()
+ val ref1 = setupEmpty("offsetpid-1")
+ val ref2 = setupEmpty("offsetpid-2")
+ ref1 ! Command("evt1-offset", probe.ref)
+ ref1 ! Command("evt2-offset", probe.ref)
+ probe.receiveMessages(2)
+ ref2 ! Command("evt3-offset", probe.ref)
+ probe.receiveMessage()
+ ref1 ! Command("evt4-offset", probe.ref)
+ probe.receiveMessage()
+
+ val all = queries.currentEventsByTag("offset",
NoOffset).runWith(Sink.seq).futureValue
+ all.map(_.event) should ===(Seq("evt1-offset", "evt2-offset",
"evt3-offset", "evt4-offset"))
+
+ queries.currentEventsByTag("offset",
all(1).offset).runWith(Sink.seq).futureValue.map(_.event) should ===(
+ Seq("evt3-offset", "evt4-offset"))
+ queries.currentEventsByTag("offset",
all.last.offset).runWith(Sink.seq).futureValue shouldBe empty
+ }
+
+ "reject unsupported offset types" in {
+ intercept[IllegalArgumentException] {
+ queries.currentEventsByTag("all", Sequence(1L))
+ }
+ }
}
}
diff --git
a/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/query/EventsBySliceSpec.scala
b/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/query/EventsBySliceSpec.scala
index 93d551198f..b8b38b6771 100644
---
a/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/query/EventsBySliceSpec.scala
+++
b/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/query/EventsBySliceSpec.scala
@@ -27,6 +27,8 @@ import pekko.actor.typed.ActorRef
import pekko.persistence.Persistence
import pekko.persistence.query.NoOffset
import pekko.persistence.query.PersistenceQuery
+import pekko.persistence.query.Sequence
+import pekko.persistence.query.TimestampOffset
import pekko.persistence.query.typed.EventEnvelope
import pekko.persistence.testkit.PersistenceTestKitPlugin
import pekko.persistence.testkit.internal.InMemStorageExtension
@@ -179,6 +181,32 @@ class EventsBySliceSpec
probe.expectNoMessage(100.millis).request(5).expectNext("e-3").expectNext("e-4")
}
+ "find events after the given offset" in {
+ val ackProbe = createTestProbe[Done]()
+ val ref = setup("o")
+ val first =
+ queries.eventsBySlices[String]("Test", 0, numberOfSlices - 1,
NoOffset).runWith(
+ TestSink[EventEnvelope[String]]()).request(2)
+ first.expectNext().offset shouldBe a[TimestampOffset]
+ val offset = first.expectNext().offset
+ first.cancel()
+
+ val probe =
+ queries.eventsBySlices[String]("Test", 0, numberOfSlices - 1,
offset).map(_.event).runWith(
+ TestSink[String]()).request(5)
+ probe.expectNext("o-3")
+ ref ! Command("o-4", ackProbe.ref)
+ ackProbe.expectMessage(Done)
+ probe.expectNext("o-4")
+ probe.cancel()
+ }
+
+ "reject unsupported offset types" in {
+ intercept[IllegalArgumentException] {
+ queries.eventsBySlices[String]("Test", 0, numberOfSlices - 1,
Sequence(1L))
+ }
+ }
+
"include timestamp in EventEnvelope" in {
setup("n")
diff --git
a/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/query/EventsByTagSpec.scala
b/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/query/EventsByTagSpec.scala
index 4bb588363a..d15a36c6f5 100644
---
a/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/query/EventsByTagSpec.scala
+++
b/persistence-testkit/src/test/scala/org/apache/pekko/persistence/testkit/query/EventsByTagSpec.scala
@@ -27,6 +27,8 @@ import pekko.actor.typed.ActorRef
import pekko.persistence.query.EventEnvelope
import pekko.persistence.query.NoOffset
import pekko.persistence.query.PersistenceQuery
+import pekko.persistence.query.Sequence
+import pekko.persistence.query.TimestampOffset
import pekko.persistence.testkit.PersistenceTestKitPlugin
import pekko.persistence.testkit.query.javadsl.{ PersistenceTestKitReadJournal
=> JavaPersistenceTestKitReadJournal }
import pekko.persistence.testkit.query.scaladsl.PersistenceTestKitReadJournal
@@ -167,6 +169,29 @@ class EventsByTagSpec
probe.expectNoMessage(100.millis).request(5).expectNext("e-3").expectNext("e-4")
}
+ "find events after the given offset" in {
+ val ackProbe = createTestProbe[Done]()
+ val tag = "offset-tag"
+ val ref = setup("offset", Set(tag))
+ val first =
queries.eventsByTag(tag).runWith(TestSink[EventEnvelope]()).request(2)
+ first.expectNext().offset shouldBe a[TimestampOffset]
+ val offset = first.expectNext().offset
+ first.cancel()
+
+ val probe = queries.eventsByTag(tag,
offset).map(_.event).runWith(TestSink[Any]()).request(5)
+ probe.expectNext("offset-3")
+ ref ! Command("offset-4", ackProbe.ref)
+ ackProbe.expectMessage(Done)
+ probe.expectNext("offset-4")
+ probe.cancel()
+ }
+
+ "reject unsupported offset types" in {
+ intercept[IllegalArgumentException] {
+ queries.eventsByTag("offset-tag", Sequence(1L))
+ }
+ }
+
"include timestamp in EventEnvelope" in {
val tag = "n-tag"
setup("n", Set(tag))
diff --git
a/persistence-typed-tests/src/test/java/org/apache/pekko/persistence/typed/javadsl/EventSourcedBehaviorJavaDslTest.java
b/persistence-typed-tests/src/test/java/org/apache/pekko/persistence/typed/javadsl/EventSourcedBehaviorJavaDslTest.java
index 03c76f1384..e0b370f8bd 100644
---
a/persistence-typed-tests/src/test/java/org/apache/pekko/persistence/typed/javadsl/EventSourcedBehaviorJavaDslTest.java
+++
b/persistence-typed-tests/src/test/java/org/apache/pekko/persistence/typed/javadsl/EventSourcedBehaviorJavaDslTest.java
@@ -15,6 +15,7 @@ package org.apache.pekko.persistence.typed.javadsl;
import static org.apache.pekko.Done.done;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonSubTypes;
@@ -41,6 +42,7 @@ import org.apache.pekko.persistence.query.EventEnvelope;
import org.apache.pekko.persistence.query.NoOffset;
import org.apache.pekko.persistence.query.PersistenceQuery;
import org.apache.pekko.persistence.query.Sequence;
+import org.apache.pekko.persistence.query.TimestampOffset;
import org.apache.pekko.persistence.testkit.PersistenceTestKitPlugin;
import org.apache.pekko.persistence.testkit.PersistenceTestKitSnapshotPlugin;
import
org.apache.pekko.persistence.testkit.query.javadsl.PersistenceTestKitReadJournal;
@@ -552,7 +554,8 @@ public class EventSourcedBehaviorJavaDslTest {
.get();
assertEquals(1, events.size());
EventEnvelope eventEnvelope = events.get(0);
- assertEquals(new Sequence(1), eventEnvelope.offset());
+ TimestampOffset offset = assertInstanceOf(TimestampOffset.class,
eventEnvelope.offset());
+ assertEquals(Map.of("tagging", 1L), offset.getSeen());
assertEquals("tagging", eventEnvelope.persistenceId());
assertEquals(new Incremented(1), eventEnvelope.event());
}
diff --git
a/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/scaladsl/EventSourcedBehaviorSpec.scala
b/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/scaladsl/EventSourcedBehaviorSpec.scala
index efc9d43093..fe1af2b6e0 100644
---
a/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/scaladsl/EventSourcedBehaviorSpec.scala
+++
b/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/scaladsl/EventSourcedBehaviorSpec.scala
@@ -13,6 +13,7 @@
package org.apache.pekko.persistence.typed.scaladsl
+import java.time.Instant
import java.util.UUID
import java.util.concurrent.atomic.AtomicInteger
@@ -42,10 +43,9 @@ import pekko.persistence.SelectedSnapshot
import pekko.persistence.journal.inmem.InmemJournal
import pekko.persistence.typed.EventAdapter
import pekko.persistence.typed.EventSeq
-import pekko.persistence.query.EventEnvelope
import pekko.persistence.query.Offset
import pekko.persistence.query.PersistenceQuery
-import pekko.persistence.query.Sequence
+import pekko.persistence.query.TimestampOffset
import pekko.persistence.snapshot.SnapshotStore
import pekko.persistence.testkit.PersistenceTestKitPlugin
import pekko.persistence.testkit.query.scaladsl.PersistenceTestKitReadJournal
@@ -58,6 +58,7 @@ import pekko.persistence.typed.SnapshotSelectionCriteria
import pekko.serialization.jackson.CborSerializable
import pekko.stream.scaladsl.Sink
+import org.scalatest.Inside.inside
import org.scalatest.wordspec.AnyWordSpecLike
import com.typesafe.config.Config
@@ -605,7 +606,12 @@ class EventSourcedBehaviorSpec
replyProbe.expectMessage(State(1, Vector(0)))
val events = queries.currentEventsByTag("tag1",
Offset.noOffset).runWith(Sink.seq).futureValue
- events shouldEqual List(EventEnvelope(Sequence(1), pid.id, 1,
Incremented(1), 0L))
+ events.map(e => (e.persistenceId, e.sequenceNr, e.event)) shouldEqual
List((pid.id, 1L, Incremented(1)))
+ events.head.timestamp should be > 0L
+ inside(events.head.offset) { case offset: TimestampOffset =>
+ offset.timestamp shouldEqual
Instant.ofEpochMilli(events.head.timestamp)
+ offset.seen shouldEqual Map(pid.id -> 1L)
+ }
}
"tag events based on state" in {
@@ -627,7 +633,11 @@ class EventSourcedBehaviorSpec
val events = queries.currentEventsByTag("higher-than-one",
Offset.noOffset).runWith(Sink.seq).futureValue
events should have size 1
- events.head shouldEqual EventEnvelope(Sequence(2), pid.id, 2,
Incremented(1), 0L)
+ val envelope = events.head
+ (envelope.persistenceId, envelope.sequenceNr, envelope.event)
shouldEqual ((pid.id, 2L, Incremented(1)))
+ inside(envelope.offset) { case offset: TimestampOffset =>
+ offset.seen shouldEqual Map(pid.id -> 2L)
+ }
}
"handle scheduled message arriving before recovery completed " in {
diff --git
a/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/scaladsl/EventSourcedEventAdapterSpec.scala
b/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/scaladsl/EventSourcedEventAdapterSpec.scala
index 46cb73542b..8a222e6bff 100644
---
a/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/scaladsl/EventSourcedEventAdapterSpec.scala
+++
b/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/scaladsl/EventSourcedEventAdapterSpec.scala
@@ -24,6 +24,7 @@ import pekko.actor.typed.scaladsl.Behaviors
import pekko.persistence.query.EventEnvelope
import pekko.persistence.query.PersistenceQuery
import pekko.persistence.query.Sequence
+import pekko.persistence.query.TimestampOffset
import pekko.persistence.testkit.PersistenceTestKitPlugin
import pekko.persistence.testkit.query.scaladsl.PersistenceTestKitReadJournal
import pekko.persistence.typed.EventAdapter
@@ -33,6 +34,7 @@ import pekko.serialization.jackson.CborSerializable
import pekko.stream.scaladsl.Sink
import pekko.testkit.JavaSerializable
+import org.scalatest.Inside.inside
import org.scalatest.wordspec.AnyWordSpecLike
import com.typesafe.config.ConfigFactory
@@ -241,7 +243,11 @@ class EventSourcedEventAdapterSpec
replyProbe.expectMessage(State(1, Vector(0)))
val taggedEvents =
queries.currentEventsByTag("tag99").runWith(Sink.seq).futureValue
- taggedEvents shouldEqual List(EventEnvelope(Sequence(1), pid.id, 1,
GenericWrapper(Incremented(1)), 0L))
+ taggedEvents.map(e => (e.persistenceId, e.sequenceNr, e.event))
shouldEqual
+ List((pid.id, 1L, GenericWrapper(Incremented(1))))
+ inside(taggedEvents.head.offset) { case offset: TimestampOffset =>
+ offset.seen shouldEqual Map(pid.id -> 1L)
+ }
}
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
