This is an automated email from the ASF dual-hosted git repository. He-Pin pushed a commit to branch fix/ignore-filtered-payload-in-replay in repository https://gitbox.apache.org/repos/asf/pekko.git
commit 8e09962c950d700e31f7e7eceee4cee44115b529 Author: 虎鸣 <[email protected]> AuthorDate: Sat Aug 15 14:38:51 2026 +0800 fix: ignore FilteredPayload in replay of persisted events Motivation: When FilteredPayload is stored in the journal (e.g. via an EventAdapter that maps certain events to FilteredPayload), replaying those events would pass FilteredPayload to the user's eventAdapter.fromJournal or receiveRecover handler, which is unexpected and can cause failures. Modification: - In ReplayingEvents (typed persistence), skip FilteredPayload by returning EmptyEventSeq before calling eventAdapter.fromJournal. - In Eventsourced (classic persistence), add a case to ignore FilteredPayload before it reaches the user's receiveRecover handler. - Add directional tests for both classic and typed persistence. Result: FilteredPayload entries in the journal are silently skipped during recovery. Sequence numbers are still correctly advanced. Tests: - sbt "persistence / Test / testOnly org.apache.pekko.persistence.PersistentActorSpec" - sbt "persistence-typed-tests / Test / testOnly org.apache.pekko.persistence.typed.scaladsl.EventSourcedBehaviorSpec" References: Refs akka/akka-core#32000 --- .../typed/scaladsl/EventSourcedBehaviorSpec.scala | 61 ++++++++++++++++++++++ .../typed/internal/ReplayingEvents.scala | 4 +- .../apache/pekko/persistence/Eventsourced.scala | 1 + .../pekko/persistence/PersistentActorSpec.scala | 34 ++++++++++++ 4 files changed, 99 insertions(+), 1 deletion(-) 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 0a65ea9645..e795d075c7 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 @@ -46,6 +46,9 @@ import pekko.persistence.query.Sequence import pekko.persistence.snapshot.SnapshotStore import pekko.persistence.testkit.PersistenceTestKitPlugin import pekko.persistence.testkit.query.scaladsl.PersistenceTestKitReadJournal +import pekko.persistence.FilteredPayload +import pekko.persistence.typed.EventAdapter +import pekko.persistence.typed.EventSeq import pekko.persistence.typed.PersistenceId import pekko.persistence.typed.RecoveryCompleted import pekko.persistence.typed.SnapshotCompleted @@ -332,6 +335,64 @@ class EventSourcedBehaviorSpec probe.expectMessage(State(4, Vector(0, 1, 2, 3))) } + "exclude FilteredPayload in replay of persisted events" in { + sealed trait Cmd + case object Incr extends Cmd + case object PersistFiltered extends Cmd + final case class GetVal(replyTo: ActorRef[Int]) extends Cmd + + sealed trait Evt + case object Incremented extends Evt + case object FilteredEvent extends Evt + + def behavior(pid: PersistenceId): EventSourcedBehavior[Cmd, Evt, Int] = + EventSourcedBehavior[Cmd, Evt, Int]( + pid, + emptyState = 0, + commandHandler = (_, cmd) => + cmd match { + case Incr => Effect.persist(Incremented) + case PersistFiltered => Effect.persist(FilteredEvent) + case GetVal(replyTo) => + Effect.none.thenRun(state => replyTo ! state) + }, + eventHandler = (state, evt) => + evt match { + case Incremented => state + 1 + case FilteredEvent => state + }) + .eventAdapter(new EventAdapter[Evt, Any] { + override def toJournal(e: Evt): Any = e match { + case FilteredEvent => FilteredPayload + case other => other + } + override def manifest(event: Evt): String = "" + override def fromJournal(p: Any, manifest: String): EventSeq[Evt] = p match { + case FilteredPayload => + throw new IllegalStateException("Unexpected FilteredPayload") + case e: Evt => EventSeq.single(e) + case other => + throw new IllegalStateException(s"Unexpected event type $other") + } + }) + + val pid = nextPid() + val c = spawn(behavior(pid)) + val probe = TestProbe[Int]() + + c ! Incr + c ! PersistFiltered + c ! Incr + c ! GetVal(probe.ref) + probe.expectMessage(2) + + // restart — recovery should skip the FilteredPayload entry + val c2 = spawn(behavior(pid)) + c2 ! Incr + c2 ! GetVal(probe.ref) + probe.expectMessage(3) + } + "handle Terminated signal" in { val c = spawn(counter(nextPid())) val probe = TestProbe[State]() diff --git a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/ReplayingEvents.scala b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/ReplayingEvents.scala index 50abf44347..e19c012b30 100644 --- a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/ReplayingEvents.scala +++ b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/ReplayingEvents.scala @@ -138,7 +138,9 @@ private[pekko] final class ReplayingEvents[C, E, S]( case ReplayedMessage(repr) => var eventForErrorReporting: OptionVal[Any] = OptionVal.None try { - val eventSeq = setup.eventAdapter.fromJournal(repr.payload, repr.manifest) + val eventSeq = + if (repr.payload == FilteredPayload) EmptyEventSeq + else setup.eventAdapter.fromJournal(repr.payload, repr.manifest) def handleEvent(event: E): Unit = { eventForErrorReporting = OptionVal.Some(event) state = state.copy(seqNr = repr.sequenceNr, eventsReplayed = state.eventsReplayed + 1) diff --git a/persistence/src/main/scala/org/apache/pekko/persistence/Eventsourced.scala b/persistence/src/main/scala/org/apache/pekko/persistence/Eventsourced.scala index ccdb90c9eb..bafafb1048 100644 --- a/persistence/src/main/scala/org/apache/pekko/persistence/Eventsourced.scala +++ b/persistence/src/main/scala/org/apache/pekko/persistence/Eventsourced.scala @@ -653,6 +653,7 @@ private[persistence] trait Eventsourced } { + case PersistentRepr(FilteredPayload, _) => // ignore case PersistentRepr(payload, _) if recoveryRunning && _receiveRecover.isDefinedAt(payload) => _receiveRecover(payload) case s: SnapshotOffer if _receiveRecover.isDefinedAt(s) => diff --git a/persistence/src/test/scala/org/apache/pekko/persistence/PersistentActorSpec.scala b/persistence/src/test/scala/org/apache/pekko/persistence/PersistentActorSpec.scala index 31f5a2238d..9ace173497 100644 --- a/persistence/src/test/scala/org/apache/pekko/persistence/PersistentActorSpec.scala +++ b/persistence/src/test/scala/org/apache/pekko/persistence/PersistentActorSpec.scala @@ -127,6 +127,24 @@ object PersistentActorSpec { extends Behavior3PersistentActor(name) with InmemRuntimePluginConfig + class Behavior4PersistentActor(name: String) extends ExamplePersistentActor(name) { + val receiveCommand: Receive = commonBehavior.orElse { + case FilteredPayload => + persist(FilteredPayload)(_ => ()) + case Cmd(data) => + persist(Evt(s"$data-${lastSequenceNr + 1}"))(updateState) + } + + override def receiveRecover: Receive = super.receiveRecover.orElse { + case FilteredPayload => + throw new IllegalStateException("Unexpected FilteredPayload") + } + } + + class Behavior4PersistentActorWithInmemRuntimePluginConfig(name: String, val providedConfig: Config) + extends Behavior4PersistentActor(name) + with InmemRuntimePluginConfig + class ChangeBehaviorInLastEventHandlerPersistentActor(name: String) extends ExamplePersistentActor(name) { val newBehavior: Receive = { case Cmd(data) => @@ -975,6 +993,8 @@ abstract class PersistentActorSpec(config: Config) extends PersistenceSpec(confi protected def behavior3PersistentActor: ActorRef = namedPersistentActor[Behavior3PersistentActor] + protected def behavior4PersistentActor: ActorRef = namedPersistentActor[Behavior4PersistentActor] + protected def changeBehaviorInFirstEventHandlerPersistentActor: ActorRef = namedPersistentActor[ChangeBehaviorInFirstEventHandlerPersistentActor] @@ -1146,6 +1166,17 @@ abstract class PersistentActorSpec(config: Config) extends PersistenceSpec(confi // cmd that was added to state before failure (b-10) is not replayed ... expectMsg(List("a-1", "a-2", "b-11", "b-12", "c-10", "c-11", "c-12")) } + "exclude FilteredPayload in replay of persisted events" in { + val persistentActor = behavior4PersistentActor + persistentActor ! GetState + expectMsg(List("a-1", "a-2")) + persistentActor ! FilteredPayload + persistentActor ! Cmd("b") + persistentActor ! "boom" + persistentActor ! Cmd("c") + persistentActor ! GetState + expectMsg(List("a-1", "a-2", "b-4", "c-5")) // seqNr 3 was for FilteredPayload + } "allow behavior changes in event handler (when handling first event)" in { val persistentActor = changeBehaviorInFirstEventHandlerPersistentActor persistentActor ! Cmd("b") @@ -1693,6 +1724,9 @@ class InmemPersistentActorWithRuntimePluginConfigSpec override protected def behavior3PersistentActor: ActorRef = namedPersistentActorWithProvidedConfig[Behavior3PersistentActorWithInmemRuntimePluginConfig](providedActorConfig) + override protected def behavior4PersistentActor: ActorRef = + namedPersistentActorWithProvidedConfig[Behavior4PersistentActorWithInmemRuntimePluginConfig](providedActorConfig) + override protected def changeBehaviorInFirstEventHandlerPersistentActor: ActorRef = namedPersistentActorWithProvidedConfig[ ChangeBehaviorInFirstEventHandlerPersistentActorWithInmemRuntimePluginConfig](providedActorConfig) --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
