ptrdom commented on code in PR #510:
URL:
https://github.com/apache/pekko-persistence-r2dbc/pull/510#discussion_r4375704243
##########
core/src/test/scala/org/apache/pekko/persistence/r2dbc/query/EventsBySliceBacktrackingSpec.scala:
##########
@@ -186,6 +201,86 @@ class EventsBySliceBacktrackingSpec
env9.sequenceNr shouldBe 4L
}
+ "mark FilteredPayload as filtered in backtracking results" in {
+ val entityType = nextEntityType()
+ val pid = nextPid(entityType)
+ val slice = query.sliceForPersistenceId(pid)
+ val sinkProbe = TestSink[EventEnvelope[String]]()(system.classicSystem)
+
+ // don't let behind-current-time be a reason for not finding events
+ val startTime = InstantFactory.now().minusSeconds(10 * 60)
+
+ val filteredSerializer =
SerializationExtension(system).findSerializerFor(FilteredPayload)
+ writeSerializedEvent(slice, pid, 1L, startTime,
filteredSerializer.identifier,
+ Serializers.manifestFor(filteredSerializer, FilteredPayload),
filteredSerializer.toBinary(FilteredPayload))
+ writeEvent(slice, pid, 2L, startTime.plusMillis(1), "e-2")
+
+ val result: TestSubscriber.Probe[EventEnvelope[String]] =
+ query
+ .eventsBySlices[String](entityType, slice, slice, NoOffset)
+ .runWith(sinkProbe)
+ .request(100)
+
+ val env1 = result.expectNext()
+ env1.sequenceNr shouldBe 1L
+ env1.source shouldBe EnvelopeOrigin.SourceQuery
+ env1.filtered shouldBe true
+ env1.eventOption shouldBe None
+
+ val env2 = result.expectNext()
+ env2.sequenceNr shouldBe 2L
+ env2.filtered shouldBe false
+ env2.eventOption shouldBe Some("e-2")
+
+ val env3 = result.expectNext()
+ env3.sequenceNr shouldBe 1L
+ env3.source shouldBe EnvelopeOrigin.SourceBacktracking
+ env3.filtered shouldBe true
+
+ val env4 = result.expectNext()
+ env4.sequenceNr shouldBe 2L
+ env4.source shouldBe EnvelopeOrigin.SourceBacktracking
+ env4.filtered shouldBe false
+
+ result.cancel()
+ }
+
+ "only mark events with the FilteredPayload manifest as filtered" in {
+ // In Pekko 2.0.0-M5 SerializedEvent shares the FilteredPayload
serializer, with manifest "S".
+ // The manifest must be checked, not only the serializer id.
+ val entityType = nextEntityType()
+ val pid = nextPid(entityType)
+ val slice = query.sliceForPersistenceId(pid)
+ val sinkProbe = TestSink[EventEnvelope[Any]]()(system.classicSystem)
+
+ // don't let behind-current-time be a reason for not finding events
+ val startTime = InstantFactory.now().minusSeconds(10 * 60)
+
+ val filteredSerializer =
SerializationExtension(system).findSerializerFor(FilteredPayload)
+ writeSerializedEvent(slice, pid, 1L, startTime,
filteredSerializer.identifier, "S", Array.emptyByteArray)
Review Comment:
This only passes on M4 because its `FilteredPayloadSerializer.fromBinary`
ignores the manifest and always returns `FilteredPayload`. In M5
(apache/pekko#3650) manifest `S` is parsed with
`MessageFormats.PersistentPayload.parseFrom`, which has required `serializerId`
and `payload` fields, so empty bytes throw. `deserializePayload(...).get` then
fails the stream, and the test fails exactly when we move to the version it is
meant to guard against.
Could you write a real payload instead? `PersistentPayload` already exists
in M4:
```scala
val stringSerializer =
SerializationExtension(system).findSerializerFor("e-1")
val bytes = MessageFormats.PersistentPayload
.newBuilder()
.setSerializerId(stringSerializer.identifier)
.setPayload(ByteString.copyFrom(stringSerializer.toBinary("e-1")))
.build()
.toByteArray
```
(`pekko.persistence.serialization.MessageFormats`,
`pekko.protobufv3.internal.ByteString`)
##########
core/src/test/scala/org/apache/pekko/persistence/r2dbc/query/EventsBySliceSpec.scala:
##########
@@ -238,6 +239,29 @@ class EventsBySliceSpec
}
}
+ "mark FilteredPayload as filtered with no payload when reading it" in
new Setup {
+ persister ! PersistWithAck(FilteredPayload, probe.ref)
Review Comment:
The event is persisted before the query starts, so in the Live variant it is
only read from the database and the PubSub path (see review summary) isn't
exercised. Could you add a Live case that subscribes first, then persists
`FilteredPayload`, and asserts `filtered == true` and `eventOption` empty? It
should fail until `PubSub.publish` is fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
