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-projection.git
The following commit(s) were added to refs/heads/main by this push:
new e2cd411b Only replay events of the stream's entity type (#633)
e2cd411b is described below
commit e2cd411b6aa4660616ead2406aefe2fe5327b888
Author: PJ Fanning <[email protected]>
AuthorDate: Sat Oct 10 18:43:14 2026 +0100
Only replay events of the stream's entity type (#633)
* Apply entity type check and producer filter to replay and loadEvent
Motivation:
Replayed events requested with a ReplayReq were not checked against the
entity type of the stream, unlike loadEvent, eventTimestamp and replay
triggered by IncludeEntityIds. Replayed events and events returned by
loadEvent also did not go through the producer filter, although the
producer filter is documented to have priority over consumer filters.
Modification:
- FilterStage ignores replay of persistence ids that don't belong to the
entity type of the stream.
- FilterStage applies the producer filter to replayed events. Events
excluded by it are emitted without payload, as FilteredEvent, so that
the consumer doesn't see a gap in sequence numbers.
- loadEvent returns FilteredEvent for events excluded by the producer
filter.
- Envelopes marked as filtered are emitted as FilteredEvent.
Result:
Replay and loadEvent are consistent with the entity type of the stream
and with the producer filter.
Tests:
- sbt grpc-test/test - 75 passed
- New FilterStageSpec and EventProducerServiceSpec tests fail without the
main code change
- sbt grpc/Compile/scalafmtCheck grpc-test/Test/scalafmtCheck - passed
- sbt grpc/headerCheckAll grpc-test/Test/headerCheckAll - passed
- MiMa not run - no mimaReportBinaryIssues task for the grpc module, only
change to internal API
- sortImports not run - task not available in this build
- grpc-int-test not run - requires database
References:
None - consistency fix for gRPC producer replay
* Keep producer filter semantics for replay and loadEvent
Motivation:
ProducerFilterEndToEndSpec showed that the producer filter is meant to
toggle whether all events of an entity are emitted. Once an event passes
the filter, the consumer replays earlier events of that entity, and those
must be emitted with payload, even when the producer filter excluded them
at the time.
Modification:
Revert the producer filter changes for replay and loadEvent, and the
related FilteredEvent handling. Keep the entity type check for replay.
Result:
Replay only emits events of the stream's entity type, and the producer
filter works as documented.
Tests:
- sbt grpc-test/test - 72 passed
- sbt grpc/Compile/scalafmtCheck grpc-test/Test/scalafmtCheck - passed
- grpc-int-test not run locally - requires Docker, left to CI
References:
None - follow-up to previous commit
* Treat malformed persistence ids in ReplayReq as not handled
Motivation:
PersistenceId.extractEntityType throws for some malformed persistence
ids, e.g. with a trailing separator, which would fail the stream.
Modification:
Wrap the entity type extraction in Try so that such persistence ids are
ignored like persistence ids of other entity types.
Result:
A ReplayReq with a malformed persistence id no longer fails the stream.
Tests:
- sbt grpc-test/testOnly FilterStageSpec - 9 passed, the added malformed
persistence id case fails without the change
References:
None - follow-up to review of this PR
---
.../projection/grpc/internal/FilterStageSpec.scala | 21 +++++++++++++++++++++
.../projection/grpc/internal/FilterStage.scala | 12 +++++++++++-
2 files changed, 32 insertions(+), 1 deletion(-)
diff --git
a/grpc-test/src/test/scala/org/apache/pekko/projection/grpc/internal/FilterStageSpec.scala
b/grpc-test/src/test/scala/org/apache/pekko/projection/grpc/internal/FilterStageSpec.scala
index 5661a097..725b572a 100644
---
a/grpc-test/src/test/scala/org/apache/pekko/projection/grpc/internal/FilterStageSpec.scala
+++
b/grpc-test/src/test/scala/org/apache/pekko/projection/grpc/internal/FilterStageSpec.scala
@@ -58,6 +58,7 @@ class FilterStageSpec extends ScalaTestWithActorTestKit("""
""") with AnyWordSpecLike with LogCapturing {
private val entityType = "EntityA"
private val streamId = "EntityAStream"
+ private val otherEntityType = "EntityB"
private val persistence = Persistence(system)
@@ -303,6 +304,26 @@ class FilterStageSpec extends ScalaTestWithActorTestKit("""
outProbe.expectNoMessage()
}
+ "ignore ReplayReq for persistence ids of other entity types or malformed
persistence ids" in new Setup {
+ override lazy val allEnvelopes = envelopes ++
+ Vector(
+ createEnvelope(PersistenceId(otherEntityType, "x"), 1, "x1"),
+ createEnvelope(ReplicationId(otherEntityType, "y",
ReplicaId("A")).persistenceId, 1, "y1"))
+
+ inPublisher.sendNext(
+ StreamIn(
+ StreamIn.Message.Replay(ReplayReq(List(
+ PersistenceIdSeqNr(PersistenceId(otherEntityType, "x").id, 1L),
+ PersistenceIdSeqNr(ReplicationId(otherEntityType, "y",
ReplicaId("A")).persistenceId.id, 1L),
+ // malformed persistence id, trailing separator
+ PersistenceIdSeqNr(s"$entityType|z|", 1L),
+ PersistenceIdSeqNr(PersistenceId(entityType, "b").id, 1L))))))
+
+ outProbe.request(10)
+ outProbe.expectNext().event shouldBe "b1"
+ outProbe.expectNoMessage()
+ }
+
"handle many replay requests" in new Setup {
lazy val entityIds = (1 to 20).map(n => s"entity-$n")
override lazy val allEnvelopes = envelopes ++
diff --git
a/grpc/src/main/scala/org/apache/pekko/projection/grpc/internal/FilterStage.scala
b/grpc/src/main/scala/org/apache/pekko/projection/grpc/internal/FilterStage.scala
index c6640c58..893fd61f 100644
---
a/grpc/src/main/scala/org/apache/pekko/projection/grpc/internal/FilterStage.scala
+++
b/grpc/src/main/scala/org/apache/pekko/projection/grpc/internal/FilterStage.scala
@@ -275,6 +275,10 @@ import org.slf4j.LoggerFactory
}
}
+ // extractEntityType may throw for malformed persistence ids, which are
then treated as not handled
+ private def entityTypeHandledByThisStream(pid: String): Boolean =
+ Try(PersistenceId.extractEntityType(pid)).toOption.contains(entityType)
+
private def sliceHandledByThisStream(pid: String): Boolean = {
val slice = persistence.sliceForPersistenceId(pid)
sliceRange.contains(slice)
@@ -327,7 +331,13 @@ import org.slf4j.LoggerFactory
private def replay(persistenceIdOffset: PersistenceIdSeqNr): Unit = {
val fromSeqNr = persistenceIdOffset.seqNr
val pid = persistenceIdOffset.persistenceId
- if (replicaIdHandledByThisStream(pid) &&
sliceHandledByThisStream(pid)) {
+ if (!entityTypeHandledByThisStream(pid)) {
+ log.debugN(
+ "Stream [{}]: Ignoring replay of persistenceId [{}], entity type
is not [{}]",
+ logPrefix,
+ pid,
+ entityType)
+ } else if (replicaIdHandledByThisStream(pid) &&
sliceHandledByThisStream(pid)) {
val sameInProgress =
replayInProgress.get(pid) match {
case Some(replay) if replay.fromSeqNr == fromSeqNr =>
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
