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 c5132df18f fix: bound persistence event replay batches (#3325)
c5132df18f is described below
commit c5132df18f383349d58a9015975238de346059a8
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Thu Jul 16 23:01:56 2026 +0800
fix: bound persistence event replay batches (#3325)
Motivation:
AsyncWriteJournal can enqueue an unbounded number of replayed events in a
recovering actor mailbox and exhaust memory.
Modification:
Add acknowledged, configurable replay batching for classic and typed
persistence while preserving replay-filter state, sender ordering, restart
isolation, and plugin compatibility.
Result:
Recovery limits in-flight replay events without truncating recovered
history.
Tests:
- Focused classic protocol, replay filter, and replay batching tests: 24
passed
- Focused typed replay batching tests: 2 passed
- sbt "persistence / Test / test": 63 passed; LevelDB JNI emitted an Apple
ARM64/x86 architecture warning
- sbt "persistence-typed-tests / Test / test": 204 passed
- sbt "persistence-shared / Test / testOnly
org.apache.pekko.persistence.journal.leveldb.PersistencePluginProxySpec": 1
passed
- sbt headerCreateAll, +headerCheckAll, checkCodeStyle, docs/paradox, and
+mimaReportBinaryIssues: passed; Paradox reported duplicate-anchor warnings
- scalafmt --list --mode diff-ref=origin/main and git diff --check: passed
- sbt sortImports: environment/task failure; the project-scoped key was
unavailable and root scalafix raised NoSuchMethodError
- sbt validatePullRequest: failed after 28m26s because LevelDB JNI 1.8 is
x86-only on macOS ARM64 and unrelated docs tests failed (UnboundedMailbox
classpath and UDP multicast interface timeouts)
References:
Fixes #3241
---
docs/src/main/paradox/persistence.md | 11 +
docs/src/main/paradox/typed/persistence.md | 11 +
.../leveldb/PersistencePluginProxySpec.scala | 14 +-
.../EventSourcedBehaviorReplayBatchingSpec.scala | 243 +++++++++++++
.../typed/internal/ExternalInteractions.scala | 3 +
.../typed/internal/ReplayingEvents.scala | 10 +
persistence/src/main/resources/reference.conf | 5 +
.../apache/pekko/persistence/Eventsourced.scala | 13 +-
.../apache/pekko/persistence/JournalProtocol.scala | 35 ++
.../persistence/journal/AsyncWriteJournal.scala | 316 ++++++++++++++---
.../journal/PersistencePluginProxy.scala | 4 +-
.../pekko/persistence/journal/ReplayFilter.scala | 10 +-
.../PersistentActorJournalProtocolSpec.scala | 9 +-
.../PersistentActorReplayBatchingSpec.scala | 387 +++++++++++++++++++++
.../persistence/journal/ReplayFilterSpec.scala | 27 ++
15 files changed, 1043 insertions(+), 55 deletions(-)
diff --git a/docs/src/main/paradox/persistence.md
b/docs/src/main/paradox/persistence.md
index 900f6d2795..36d0f92514 100644
--- a/docs/src/main/paradox/persistence.md
+++ b/docs/src/main/paradox/persistence.md
@@ -136,6 +136,17 @@ until other recoveries have been completed. This is
configured by:
pekko.persistence.max-concurrent-recoveries = 50
```
+Journal plugins based on `AsyncWriteJournal` also bound each individual
recovery. They replay at most
+`replay-batch-size` event sequence numbers and wait until the recovering actor
has processed that batch before
+requesting the next one. The default is `1000`; it can be changed in the
journal plugin configuration, for example:
+
+```
+pekko.persistence.journal.my-plugin.replay-batch-size = 500
+```
+
+This setting limits replay pressure on the persistent actor's mailbox without
limiting the total number of events
+that are recovered.
+
@@@ note
Accessing the @scala[`sender()`]@java[sender with `getSender()`] for replayed
messages will always result in a `deadLetters` reference,
diff --git a/docs/src/main/paradox/typed/persistence.md
b/docs/src/main/paradox/typed/persistence.md
index 3e4c2aea65..172067f0bb 100644
--- a/docs/src/main/paradox/typed/persistence.md
+++ b/docs/src/main/paradox/typed/persistence.md
@@ -468,6 +468,17 @@ until other recoveries have been completed. This is
configured by:
pekko.persistence.max-concurrent-recoveries = 50
```
+Journal plugins based on `AsyncWriteJournal` also bound each individual
recovery. They replay at most
+`replay-batch-size` event sequence numbers and wait until the recovering actor
has processed that batch before
+requesting the next one. The default is `1000`; it can be changed in the
journal plugin configuration, for example:
+
+```
+pekko.persistence.journal.my-plugin.replay-batch-size = 500
+```
+
+This setting limits replay pressure on the persistent actor's mailbox without
limiting the total number of events
+that are recovered.
+
The @ref:[event handler](#event-handler) is used for updating the state when
replaying the journaled events.
It is strongly discouraged to perform side effects in the event handler, so
side effects should be performed
diff --git
a/persistence-shared/src/test/scala/org/apache/pekko/persistence/journal/leveldb/PersistencePluginProxySpec.scala
b/persistence-shared/src/test/scala/org/apache/pekko/persistence/journal/leveldb/PersistencePluginProxySpec.scala
index 3364cfef33..01baf79c63 100644
---
a/persistence-shared/src/test/scala/org/apache/pekko/persistence/journal/leveldb/PersistencePluginProxySpec.scala
+++
b/persistence-shared/src/test/scala/org/apache/pekko/persistence/journal/leveldb/PersistencePluginProxySpec.scala
@@ -32,6 +32,10 @@ object PersistencePluginProxySpec {
journal {
plugin = "pekko.persistence.journal.proxy"
proxy.target-journal-plugin = "pekko.persistence.journal.inmem"
+ inmem {
+ replay-batch-size = 2
+ replay-filter.mode = off
+ }
}
snapshot-store {
plugin = "pekko.persistence.snapshot-store.proxy"
@@ -131,20 +135,20 @@ class PersistencePluginProxySpec
val appA = systemA.actorOf(Props(classOf[ExampleApp], probeA.ref))
val appB = systemB.actorOf(Props(classOf[ExampleApp], probeB.ref))
- appA ! "a1"
+ val eventsA = Vector("a1", "a2", "a3", "a4", "a5")
+ eventsA.foreach(appA ! _)
appB ! "b1"
- probeA.expectMsg("a1")
+ eventsA.foreach(event => probeA.expectMsg(event))
probeB.expectMsg("b1")
val recoveredAppA = systemA.actorOf(Props(classOf[ExampleApp],
probeA.ref))
val recoveredAppB = systemB.actorOf(Props(classOf[ExampleApp],
probeB.ref))
- recoveredAppA ! "a2"
+ recoveredAppA ! "a6"
recoveredAppB ! "b2"
- probeA.expectMsg("a1")
- probeA.expectMsg("a2")
+ (eventsA :+ "a6").foreach(event => probeA.expectMsg(event))
probeB.expectMsg("b1")
probeB.expectMsg("b2")
diff --git
a/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/scaladsl/EventSourcedBehaviorReplayBatchingSpec.scala
b/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/scaladsl/EventSourcedBehaviorReplayBatchingSpec.scala
new file mode 100644
index 0000000000..ba8b0726d0
--- /dev/null
+++
b/persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/scaladsl/EventSourcedBehaviorReplayBatchingSpec.scala
@@ -0,0 +1,243 @@
+/*
+ * 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.typed.scaladsl
+
+import scala.concurrent.{ Await, Future, Promise }
+import scala.concurrent.duration._
+
+import org.apache.pekko
+import pekko.actor.Actor
+import pekko.actor.testkit.typed.scaladsl.{ LogCapturing,
ScalaTestWithActorTestKit }
+import pekko.actor.typed.{ ActorRef, Behavior, PreRestart, SupervisorStrategy }
+import pekko.persistence.JournalProtocol.{ ReplayBatchResponse,
ReplayedMessage }
+import pekko.persistence.PersistentRepr
+import pekko.persistence.journal.SteppingInmemJournal
+import pekko.persistence.journal.inmem.InmemJournal
+import pekko.persistence.typed.{ PersistenceId, RecoveryCompleted,
RecoveryFailed }
+import pekko.persistence.typed.internal.EventSourcedBehaviorImpl.WriterIdentity
+
+import org.scalatest.wordspec.AnyWordSpecLike
+
+import com.typesafe.config.ConfigFactory
+
+object DelayedReplayInmemJournal {
+ private var plannedInvocations = 0
+ private var nextInvocation = 0
+ private var started = Vector.empty[Promise[Unit]]
+ private var completions = Map.empty[Int, () => Unit]
+
+ def prepare(invocations: Int): Vector[Future[Unit]] = synchronized {
+ require(nextInvocation == plannedInvocations && completions.isEmpty,
"previous delayed replay is still active")
+ plannedInvocations = invocations
+ nextInvocation = 0
+ started = Vector.fill(invocations)(Promise[Unit]())
+ started.map(_.future)
+ }
+
+ private def claim(): Option[Int] = synchronized {
+ if (nextInvocation == plannedInvocations) None
+ else {
+ val invocation = nextInvocation
+ nextInvocation += 1
+ Some(invocation)
+ }
+ }
+
+ private def register(invocation: Int, completion: () => Unit): Unit =
synchronized {
+ completions = completions.updated(invocation, completion)
+ started(invocation).success(())
+ }
+
+ def complete(invocation: Int): Unit = {
+ val completion = synchronized {
+ val result = completions.getOrElse(invocation, throw new
IllegalStateException("delayed replay has not started"))
+ completions -= invocation
+ result
+ }
+ completion()
+ }
+}
+
+final class DelayedReplayInmemJournal extends InmemJournal {
+ import DelayedReplayInmemJournal._
+ import context.dispatcher
+
+ override def asyncReplayMessages(persistenceId: String, fromSequenceNr:
Long, toSequenceNr: Long, max: Long)(
+ recoveryCallback: PersistentRepr => Unit): Future[Unit] =
+ claim() match {
+ case None => super.asyncReplayMessages(persistenceId,
fromSequenceNr, toSequenceNr, max)(recoveryCallback)
+ case Some(invocation) =>
+ val buffered = Vector.newBuilder[PersistentRepr]
+ super
+ .asyncReplayMessages(persistenceId, fromSequenceNr, toSequenceNr,
max)(buffered += _)
+ .flatMap { _ =>
+ val promise = Promise[Unit]()
+ val replayed = buffered.result()
+ register(
+ invocation,
+ () => {
+ replayed.foreach(recoveryCallback)
+ promise.success(())
+ })
+ promise.future
+ }
+ }
+}
+
+object EventSourcedBehaviorReplayBatchingSpec {
+ val JournalId = "event-sourced-behavior-replay-batching-spec"
+
+ sealed trait Command
+ final case class PersistAll(events: Vector[String]) extends Command
+ final case class Applied(event: String)
+ final case class RecoveryFinished(state: Vector[String])
+ case object RecoveryFailedObserved
+ case object Restarting
+
+ val config =
+ SteppingInmemJournal
+ .config(JournalId)
+ .withFallback(ConfigFactory.parseString(s"""
+ pekko.persistence.journal.stepping-inmem {
+ replay-batch-size = 2
+ replay-filter.mode = off
+ }
+ delayed-replay-journal = $${pekko.persistence.journal.inmem}
+ delayed-replay-journal {
+ class =
"org.apache.pekko.persistence.typed.scaladsl.DelayedReplayInmemJournal"
+ replay-batch-size = 2
+ replay-filter.mode = off
+ recovery-event-timeout = 1s
+ }
+ """)).withFallback(ConfigFactory.defaultReference()).resolve()
+
+ def behavior(persistenceId: PersistenceId, probe: ActorRef[AnyRef]):
Behavior[Command] =
+ EventSourcedBehavior[Command, String, Vector[String]](
+ persistenceId,
+ emptyState = Vector.empty,
+ commandHandler = (_, command) =>
+ command match {
+ case PersistAll(events) => Effect.persist(events)
+ },
+ eventHandler = (state, event) => {
+ probe ! Applied(event)
+ state :+ event
+ }).receiveSignal {
+ case (state, RecoveryCompleted) => probe ! RecoveryFinished(state)
+ }
+
+ def restartingBehavior(persistenceId: PersistenceId, probe:
ActorRef[AnyRef]): Behavior[Command] =
+ EventSourcedBehavior[Command, String, Vector[String]](
+ persistenceId,
+ emptyState = Vector.empty,
+ commandHandler = (_, command) =>
+ command match {
+ case PersistAll(events) => Effect.persist(events)
+ },
+ eventHandler = (state, event) => {
+ probe ! Applied(event)
+ state :+ event
+ }).withJournalPluginId("delayed-replay-journal").receiveSignal {
+ case (state, RecoveryCompleted) => probe ! RecoveryFinished(state)
+ case (_, RecoveryFailed(_)) => probe ! RecoveryFailedObserved
+ case (_, PreRestart) => probe ! Restarting
+ }.onPersistFailure(
+ SupervisorStrategy.restartWithBackoff(10.millis, 10.millis, randomFactor
= 0.0).withLoggingEnabled(false))
+}
+
+class EventSourcedBehaviorReplayBatchingSpec
+ extends
ScalaTestWithActorTestKit(EventSourcedBehaviorReplayBatchingSpec.config)
+ with AnyWordSpecLike
+ with LogCapturing {
+ import EventSourcedBehaviorReplayBatchingSpec._
+
+ import org.apache.pekko.actor.typed.scaladsl.adapter._
+ private implicit val classicSystem: pekko.actor.ActorSystem =
system.toClassic
+
+ "An EventSourcedBehavior recovery" must {
+ "acknowledge each bounded replay batch before the journal produces the
next batch" in {
+ val probe = createTestProbe[AnyRef]()
+ val persistenceId = PersistenceId.ofUniqueId("bounded-replay")
+ val events = Vector("a", "b", "c", "d", "e")
+
+ val persisting = spawn(behavior(persistenceId, probe.ref))
+ probe.awaitAssert(SteppingInmemJournal.getRef(JournalId), 3.seconds)
+ val journal = SteppingInmemJournal.getRef(JournalId)
+
+ SteppingInmemJournal.step(journal)
+ probe.expectMessage(RecoveryFinished(Vector.empty))
+
+ persisting ! PersistAll(events)
+ SteppingInmemJournal.step(journal)
+ events.foreach(event => probe.expectMessage(Applied(event)))
+
+ testKit.stop(persisting)
+ probe.expectTerminated(persisting)
+
+ spawn(behavior(persistenceId, probe.ref))
+ SteppingInmemJournal.step(journal) // read highest sequence number
+
+ SteppingInmemJournal.step(journal) // replay 1-2
+ probe.expectMessage(Applied("a"))
+ probe.expectMessage(Applied("b"))
+ probe.expectNoMessage(100.millis)
+
+ SteppingInmemJournal.step(journal) // replay 3-4
+ probe.expectMessage(Applied("c"))
+ probe.expectMessage(Applied("d"))
+ probe.expectNoMessage(100.millis)
+
+ SteppingInmemJournal.step(journal) // replay 5 and complete
+ probe.expectMessage(Applied("e"))
+ probe.expectMessage(RecoveryFinished(events))
+ }
+
+ "isolate a timed-out replay from a new recovery using the same actor
reference" in {
+ val probe = createTestProbe[AnyRef]()
+ val persistenceId = PersistenceId.ofUniqueId("timed-out-replay")
+ val events = Vector("a", "b", "c")
+
+ val persisting = spawn(restartingBehavior(persistenceId, probe.ref))
+ probe.expectMessage(RecoveryFinished(Vector.empty))
+ persisting ! PersistAll(events)
+ events.foreach(event => probe.expectMessage(Applied(event)))
+ testKit.stop(persisting)
+ probe.expectTerminated(persisting)
+
+ val replayStarted = DelayedReplayInmemJournal.prepare(invocations = 2)
+ val recovering = spawn(restartingBehavior(persistenceId, probe.ref))
+ Await.result(replayStarted.head, 3.seconds)
+ val timedOutActorInstanceId = WriterIdentity.instanceIdCounter.get() - 1
+ probe.expectMessageType[RecoveryFailedObserved.type](5.seconds)
+ probe.expectMessage(Restarting)
+ Await.result(replayStarted(1), 3.seconds)
+
+ recovering.toClassic.tell(
+ ReplayBatchResponse(
+ timedOutActorInstanceId,
+ ReplayedMessage(PersistentRepr("stale", sequenceNr = 1L,
persistenceId = persistenceId.id))),
+ Actor.noSender)
+ DelayedReplayInmemJournal.complete(0)
+ probe.expectNoMessage(100.millis)
+
+ DelayedReplayInmemJournal.complete(1)
+ events.foreach(event => probe.expectMessage(Applied(event)))
+ probe.expectMessage(RecoveryFinished(events))
+ }
+ }
+}
diff --git
a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/ExternalInteractions.scala
b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/ExternalInteractions.scala
index 10f8b6e497..134fe4c4eb 100644
---
a/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/ExternalInteractions.scala
+++
b/persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/ExternalInteractions.scala
@@ -135,6 +135,9 @@ private[pekko] trait JournalInteractions[C, E, S] {
protected def replayEvents(fromSeqNr: Long, toSeqNr: Long): Unit = {
setup.internalLogger.debug2("Replaying events: from: {}, to: {}",
fromSeqNr, toSeqNr)
+ setup.journal.tell(
+
JournalProtocol.ReplayMessagesWithBatching(setup.writerIdentity.instanceId),
+ setup.selfClassic)
setup.journal.tell(
ReplayMessages(fromSeqNr, toSeqNr, setup.recovery.replayMax,
setup.persistenceId.id, setup.selfClassic),
setup.selfClassic)
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 db902d9bb6..50abf44347 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
@@ -131,6 +131,10 @@ private[pekko] final class ReplayingEvents[C, E, S](
private def onJournalResponse(response: JournalProtocol.Response):
Behavior[InternalProtocol] = {
try {
response match {
+ case ReplayBatchResponse(actorInstanceId, currentResponse) =>
+ if (actorInstanceId == setup.writerIdentity.instanceId)
onJournalResponse(currentResponse)
+ else this
+
case ReplayedMessage(repr) =>
var eventForErrorReporting: OptionVal[Any] = OptionVal.None
try {
@@ -192,6 +196,11 @@ private[pekko] final class ReplayingEvents[C, E, S](
onRecoveryFailure(ex, eventForErrorReporting.toOption)
}
+ case ReplayBatchReady(replayId) =>
+ state = state.copy(eventSeenInInterval = true)
+ setup.journal.tell(ReplayBatchAck(replayId), setup.selfClassic)
+ this
+
case RecoverySuccess(highestJournalSeqNr) =>
val highestSeqNr = Math.max(highestJournalSeqNr, state.seqNr)
state = state.copy(seqNr = highestSeqNr)
@@ -255,6 +264,7 @@ private[pekko] final class ReplayingEvents[C, E, S](
* @param event the event that was being processed when the exception was
thrown
*/
private def onRecoveryFailure(cause: Throwable, event: Option[Any]):
Behavior[InternalProtocol] = {
+ setup.journal.tell(ReplayMessagesCancel, setup.selfClassic)
onRecoveryFailed(setup.context, cause, event)
setup.onSignal(state.state, RecoveryFailed(cause), catchAndLog = true)
setup.cancelRecoveryTimer()
diff --git a/persistence/src/main/resources/reference.conf
b/persistence/src/main/resources/reference.conf
index ff34a035bb..c00b1f5763 100644
--- a/persistence/src/main/resources/reference.conf
+++ b/persistence/src/main/resources/reference.conf
@@ -115,6 +115,11 @@ pekko.persistence {
# as it has accumulated since the last write.
max-message-batch-size = 200
+ # Maximum number of events that an AsyncWriteJournal replays before
waiting for
+ # the recovering persistent actor to acknowledge that it has processed
the batch.
+ # This bounds the number of replayed events queued in the persistent
actor's mailbox.
+ replay-batch-size = 1000
+
# If there is more time in between individual events gotten from the
journal
# recovery than this the recovery will fail.
# Note that it also affects reading the snapshot before replaying events
on
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 cb2006526c..e0593256ab 100644
--- a/persistence/src/main/scala/org/apache/pekko/persistence/Eventsourced.scala
+++ b/persistence/src/main/scala/org/apache/pekko/persistence/Eventsourced.scala
@@ -253,7 +253,12 @@ private[persistence] trait Eventsourced
/** INTERNAL API. */
override protected[pekko] def aroundReceive(receive: Receive, message: Any):
Unit =
- currentState.stateReceive(receive, message)
+ message match {
+ case ReplayBatchResponse(`instanceId`, response) if
currentState.recoveryRunning =>
+ currentState.stateReceive(receive, response)
+ case _: ReplayBatchResponse => // stale or late replay response
+ case other => currentState.stateReceive(receive, other)
+ }
/** INTERNAL API. */
override protected[pekko] def aroundPreStart(): Unit = {
@@ -684,6 +689,7 @@ private[persistence] trait Eventsourced
}
}
changeState(recovering(recoveryBehavior, timeout))
+ journal ! ReplayMessagesWithBatching(instanceId)
journal ! ReplayMessages(lastSequenceNr + 1L, toSnr, replayMax,
persistenceId, self)
}
@@ -762,7 +768,7 @@ private[persistence] trait Eventsourced
override def recoveryRunning: Boolean = _recoveryRunning
- override def stateReceive(receive: Receive, message: Any) =
+ override def stateReceive(receive: Receive, message: Any): Unit =
try message match {
case ReplayedMessage(p) =>
try {
@@ -776,6 +782,9 @@ private[persistence] trait Eventsourced
finally context.stop(self)
returnRecoveryPermit()
}
+ case ReplayBatchReady(replayId) =>
+ eventSeenInInterval = true
+ journal ! ReplayBatchAck(replayId)
case RecoverySuccess(highestJournalSeqNr) =>
timeoutCancellable.cancel()
onReplaySuccess() // callback for subclass implementation
diff --git
a/persistence/src/main/scala/org/apache/pekko/persistence/JournalProtocol.scala
b/persistence/src/main/scala/org/apache/pekko/persistence/JournalProtocol.scala
index d82b427906..c23ecfc469 100644
---
a/persistence/src/main/scala/org/apache/pekko/persistence/JournalProtocol.scala
+++
b/persistence/src/main/scala/org/apache/pekko/persistence/JournalProtocol.scala
@@ -125,6 +125,26 @@ private[persistence] object JournalProtocol {
persistentActor: ActorRef)
extends Request
+ /**
+ * Opts the sender in to bounded replay for the immediately following
[[ReplayMessages]] request.
+ * Journals that don't support bounded replay can ignore this message and
process [[ReplayMessages]] as before.
+ * The `actorInstanceId` identifies the current incarnation of the
recovering actor so that responses from a
+ * cancelled replay can be discarded after a restart.
+ */
+ final case class ReplayMessagesWithBatching(actorInstanceId: Int) extends
Request
+
+ /** Cancels all bounded replays requested by the sender. */
+ case object ReplayMessagesCancel extends Request
+
+ /**
+ * Acknowledges that the requester has processed the replayed messages
emitted before the corresponding
+ * [[ReplayBatchReady]].
+ */
+ final case class ReplayBatchAck(replayId: Long) extends Request
+
+ /** Cancels a bounded replay after its replay filter has failed. */
+ final case class ReplayBatchCancel(replayId: Long) extends Request
+
/**
* Reply message to a [[ReplayMessages]] request. A separate reply is sent
to the requester for each
* replayed message.
@@ -136,6 +156,21 @@ private[persistence] object JournalProtocol {
with DeadLetterSuppression
with NoSerializationVerificationNeeded
+ /**
+ * Signals that one bounded replay batch has been emitted. The requester
must process all preceding
+ * [[ReplayedMessage]] responses and reply with [[ReplayBatchAck]] before
the journal emits another batch.
+ */
+ final case class ReplayBatchReady(replayId: Long)
+ extends Response
+ with DeadLetterSuppression
+ with NoSerializationVerificationNeeded
+
+ /** Associates a bounded replay response with the recovering actor
incarnation that requested it. */
+ final case class ReplayBatchResponse(actorInstanceId: Int, response:
Response)
+ extends Response
+ with DeadLetterSuppression
+ with NoSerializationVerificationNeeded
+
/**
* Reply message to a successful [[ReplayMessages]] request. This reply is
sent to the requester
* after all [[ReplayedMessage]] have been sent (if any).
diff --git
a/persistence/src/main/scala/org/apache/pekko/persistence/journal/AsyncWriteJournal.scala
b/persistence/src/main/scala/org/apache/pekko/persistence/journal/AsyncWriteJournal.scala
index ae27bd5c4f..6afe37527e 100644
---
a/persistence/src/main/scala/org/apache/pekko/persistence/journal/AsyncWriteJournal.scala
+++
b/persistence/src/main/scala/org/apache/pekko/persistence/journal/AsyncWriteJournal.scala
@@ -13,7 +13,9 @@
package org.apache.pekko.persistence.journal
-import scala.collection.immutable
+import java.util.concurrent.atomic.AtomicLong
+
+import scala.collection.{ immutable, mutable }
import scala.concurrent.ExecutionContext
import scala.concurrent.Future
import scala.concurrent.duration._
@@ -69,10 +71,138 @@ trait AsyncWriteJournal extends Actor with
WriteJournalBase with AsyncRecovery {
// cannot be a val in the trait due to binary compatibility
val replayDebugEnabled: Boolean = config.getBoolean("replay-filter.debug")
val enableGlobalWriteResponseOrder: Boolean =
config.getBoolean("write-response-global-order")
+ val replayBatchSize: Long = config.getLong("replay-batch-size")
+ require(replayBatchSize > 0L, s"replay-batch-size must be greater than 0,
but was [$replayBatchSize]")
val eventStream = context.system.eventStream // used from Future callbacks
implicit val ec: ExecutionContext = context.dispatcher
+ val batchingReplayRequesters = mutable.Map.empty[ActorRef, Int]
+ val replaySessions = mutable.Map.empty[Long, ReplaySession]
+
+ def replayTarget(persistentActor: ActorRef): ActorRef =
+ if (isReplayFilterEnabled)
+ context.actorOf(
+ ReplayFilter.props(
+ persistentActor,
+ replayFilterMode,
+ replayFilterWindowSize,
+ replayFilterMaxOldWriters,
+ replayDebugEnabled))
+ else persistentActor
+
+ def batchedReplayTarget(persistentActor: ActorRef, actorInstanceId: Int):
(ActorRef, ActorRef) = {
+ val relay = context.actorOf(replayForwarderProps(persistentActor,
actorInstanceId))
+ val replyTo =
+ if (isReplayFilterEnabled)
+ context.actorOf(
+ ReplayFilter.props(
+ relay,
+ replayFilterMode,
+ replayFilterWindowSize,
+ replayFilterMaxOldWriters,
+ replayDebugEnabled))
+ else relay
+ (replyTo, relay)
+ }
+
+ def hasReplaySession(persistentActor: ActorRef): Boolean =
+ replaySessions.valuesIterator.exists(_.request.persistentActor ==
persistentActor)
+
+ def unwatchIfUnused(persistentActor: ActorRef): Unit =
+ if (!batchingReplayRequesters.contains(persistentActor) &&
!hasReplaySession(persistentActor))
+ context.unwatch(persistentActor)
+
+ def removeReplaySession(replayId: Long): Option[ReplaySession] =
+ replaySessions.remove(replayId).map { session =>
+ val persistentActor = session.request.persistentActor
+ unwatchIfUnused(persistentActor)
+ session
+ }
+
+ def publishReplayRequest(session: ReplaySession): Unit =
+ if (publish) eventStream.publish(session.request)
+
+ def finishReplay(replayId: Long, response: JournalProtocol.Response): Unit
=
+ removeReplaySession(replayId).foreach { session =>
+ session.replyTo.tell(response, self)
+ publishReplayRequest(session)
+ }
+
+ def cancelReplay(replayId: Long): Unit =
+ removeReplaySession(replayId).foreach { session =>
+ if (session.replyTo != session.relay)
+ context.stop(session.replyTo)
+ context.stop(session.relay)
+ publishReplayRequest(session)
+ }
+
+ def cancelReplaysFor(persistentActor: ActorRef): Unit =
+ replaySessions.valuesIterator
+ .filter(_.request.persistentActor == persistentActor)
+ .map(_.replayId)
+ .toVector
+ .foreach(cancelReplay)
+
+ def batchUpperBound(fromSequenceNr: Long, toSequenceNr: Long): Long = {
+ val increment = replayBatchSize - 1L
+ val upperBound =
+ if (fromSequenceNr > Long.MaxValue - increment) Long.MaxValue
+ else fromSequenceNr + increment
+ math.min(toSequenceNr, upperBound)
+ }
+
+ def startReplayBatch(session: ReplaySession): Unit = {
+ val batchToSequenceNr = batchUpperBound(session.nextSequenceNr,
session.toSequenceNr)
+ val batchMax = math.min(replayBatchSize, session.remaining)
+ val replayed = new AtomicLong
+ val replayResult =
+ try
+ asyncReplayMessages(
+ session.request.persistenceId,
+ session.nextSequenceNr,
+ batchToSequenceNr,
+ batchMax) { p =>
+ replayed.incrementAndGet()
+ if (!p.deleted) // old records from Akka 2.3 may still have the
deleted flag
+ adaptFromJournal(p).foreach { adaptedPersistentRepr =>
+ session.replyTo.tell(ReplayedMessage(adaptedPersistentRepr),
self)
+ }
+ }
+ catch { case NonFatal(e) => Future.failed(e) }
+
+ replayResult
+ .map(_ => ReplayBatchCompleted(session.replayId, batchToSequenceNr,
replayed.get()))
+ .recover { case e => ReplayFailed(session.replayId, e) }
+ .pipeTo(self)
+ }
+
+ def startBatchedReplay(request: ReplayMessages, actorInstanceId: Int):
Unit = {
+ val replayId = nextReplayId()
+ val (replyTo, relay) = batchedReplayTarget(request.persistentActor,
actorInstanceId)
+ val session = ReplaySession(
+ replayId,
+ request,
+ replyTo,
+ relay,
+ highestSequenceNr = 0L,
+ toSequenceNr = 0L,
+ nextSequenceNr = request.fromSequenceNr,
+ remaining = request.max,
+ awaitingAck = false)
+ replaySessions.put(replayId, session)
+ context.watch(request.persistentActor)
+
+ val readHighestSequenceNrFrom = math.max(0L, request.fromSequenceNr - 1L)
+ val readHighestResult =
+ try
breaker.withCircuitBreaker(asyncReadHighestSequenceNr(request.persistenceId,
readHighestSequenceNrFrom))
+ catch { case NonFatal(e) => Future.failed(e) }
+ readHighestResult
+ .map(ReplayHighestSequenceNr(replayId, _))
+ .recover { case e => ReplayFailed(replayId, e) }
+ .pipeTo(self)
+ }
+
// should be a private method in the trait, but it needs the
enableGlobalWriteResponseOrder field which can't be
// moved to the trait level because adding any fields there breaks
bincompat
@InternalApi
@@ -159,52 +289,113 @@ trait AsyncWriteJournal extends Actor with
WriteJournalBase with AsyncRecovery {
}
}
+ case ReplayMessagesWithBatching(actorInstanceId) =>
+ val requester = sender()
+ cancelReplaysFor(requester)
+ batchingReplayRequesters.put(requester, actorInstanceId)
+ context.watch(requester)
+
+ case ReplayMessagesCancel =>
+ val requester = sender()
+ batchingReplayRequesters.remove(requester)
+ cancelReplaysFor(requester)
+ unwatchIfUnused(requester)
+
case r @ ReplayMessages(fromSequenceNr, toSequenceNr, max,
persistenceId, persistentActor) =>
- val replyTo =
- if (isReplayFilterEnabled)
- context.actorOf(
- ReplayFilter.props(
- persistentActor,
- replayFilterMode,
- replayFilterWindowSize,
- replayFilterMaxOldWriters,
- replayDebugEnabled))
- else persistentActor
-
- val readHighestSequenceNrFrom = math.max(0L, fromSequenceNr - 1)
- /*
- * The API docs for the [[AsyncRecovery]] say not to rely on
asyncReadHighestSequenceNr
- * being called before a call to asyncReplayMessages even tho it
currently always is. The Cassandra
- * plugin does rely on this so if you change this change the Cassandra
plugin.
- */
- breaker
- .withCircuitBreaker(asyncReadHighestSequenceNr(persistenceId,
readHighestSequenceNrFrom))
- .flatMap { highSeqNr =>
- val toSeqNr = math.min(toSequenceNr, highSeqNr)
- if (toSeqNr <= 0L || fromSequenceNr > toSeqNr)
- Future.successful(highSeqNr)
- else {
- // Send replayed messages and replay result to persistentActor
directly. No need
- // to resequence replayed messages relative to written and
looped messages.
- // not possible to use circuit breaker here
- asyncReplayMessages(persistenceId, fromSequenceNr, toSeqNr, max)
{ p =>
- if (!p.deleted) // old records from Akka 2.3 may still have
the deleted flag
- adaptFromJournal(p).foreach { adaptedPersistentRepr =>
- replyTo.tell(ReplayedMessage(adaptedPersistentRepr),
Actor.noSender)
- }
- }.map(_ => highSeqNr)
+ val requester = sender()
+ if (requester == persistentActor)
+ cancelReplaysFor(persistentActor)
+ val requestedBatching = batchingReplayRequesters.remove(requester)
+ if (requestedBatching.isDefined && requester == persistentActor) {
+ startBatchedReplay(r, requestedBatching.get)
+ } else {
+ if (requestedBatching.isDefined)
+ unwatchIfUnused(requester)
+ val replyTo = replayTarget(persistentActor)
+
+ val readHighestSequenceNrFrom = math.max(0L, fromSequenceNr - 1)
+ /*
+ * The API docs for the [[AsyncRecovery]] say not to rely on
asyncReadHighestSequenceNr
+ * being called before a call to asyncReplayMessages even tho it
currently always is. The Cassandra
+ * plugin does rely on this so if you change this change the
Cassandra plugin.
+ */
+ breaker
+ .withCircuitBreaker(asyncReadHighestSequenceNr(persistenceId,
readHighestSequenceNrFrom))
+ .flatMap { highSeqNr =>
+ val toSeqNr = math.min(toSequenceNr, highSeqNr)
+ if (toSeqNr <= 0L || fromSequenceNr > toSeqNr)
+ Future.successful(highSeqNr)
+ else {
+ // Send replayed messages and replay result to persistentActor
directly. No need
+ // to resequence replayed messages relative to written and
looped messages.
+ // not possible to use circuit breaker here
+ asyncReplayMessages(persistenceId, fromSequenceNr, toSeqNr,
max) { p =>
+ if (!p.deleted) // old records from Akka 2.3 may still have
the deleted flag
+ adaptFromJournal(p).foreach { adaptedPersistentRepr =>
+ replyTo.tell(ReplayedMessage(adaptedPersistentRepr),
Actor.noSender)
+ }
+ }.map(_ => highSeqNr)
+ }
}
+ .map { highSeqNr =>
+ RecoverySuccess(highSeqNr)
+ }
+ .recover {
+ case e => ReplayMessagesFailure(e)
+ }
+ .pipeTo(replyTo)
+ .foreach { _ =>
+ if (publish) eventStream.publish(r)
+ }
+ }
+
+ case ReplayHighestSequenceNr(replayId, highSeqNr) =>
+ replaySessions.get(replayId).foreach { session =>
+ val toSeqNr = math.min(session.request.toSequenceNr, highSeqNr)
+ val initialized = session.copy(highestSequenceNr = highSeqNr,
toSequenceNr = toSeqNr)
+ replaySessions.update(replayId, initialized)
+ if (toSeqNr <= 0L || session.nextSequenceNr > toSeqNr ||
session.remaining <= 0L)
+ finishReplay(replayId, RecoverySuccess(highSeqNr))
+ else
+ startReplayBatch(initialized)
+ }
+
+ case ReplayBatchCompleted(replayId, batchToSequenceNr, replayed) =>
+ replaySessions.get(replayId).foreach { session =>
+ val remaining = if (replayed >= session.remaining) 0L else
session.remaining - replayed
+ if (remaining == 0L || batchToSequenceNr >= session.toSequenceNr) {
+ finishReplay(replayId, RecoverySuccess(session.highestSequenceNr))
+ } else {
+ val waiting = session.copy(
+ nextSequenceNr = batchToSequenceNr + 1L,
+ remaining = remaining,
+ awaitingAck = true)
+ replaySessions.update(replayId, waiting)
+ session.replyTo.tell(ReplayBatchReady(replayId), self)
}
- .map { highSeqNr =>
- RecoverySuccess(highSeqNr)
- }
- .recover {
- case e => ReplayMessagesFailure(e)
- }
- .pipeTo(replyTo)
- .foreach { _ =>
- if (publish) eventStream.publish(r)
- }
+ }
+
+ case ReplayBatchAck(replayId) =>
+ replaySessions.get(replayId) match {
+ case Some(session) if session.awaitingAck && sender() ==
session.request.persistentActor =>
+ val replaying = session.copy(awaitingAck = false)
+ replaySessions.update(replayId, replaying)
+ startReplayBatch(replaying)
+ case _ => // stale or invalid acknowledgement
+ }
+
+ case ReplayBatchCancel(replayId) =>
+ replaySessions.get(replayId) match {
+ case Some(session) if sender() == session.replyTo =>
cancelReplay(replayId)
+ case _ => // stale or
invalid cancellation
+ }
+
+ case ReplayFailed(replayId, cause) =>
+ finishReplay(replayId, ReplayMessagesFailure(cause))
+
+ case Terminated(persistentActor) =>
+ batchingReplayRequesters.remove(persistentActor)
+ cancelReplaysFor(persistentActor)
case d @ DeleteMessagesTo(persistenceId, toSequenceNr, persistentActor)
=>
breaker
@@ -317,8 +508,45 @@ trait AsyncWriteJournal extends Actor with
WriteJournalBase with AsyncRecovery {
* INTERNAL API.
*/
private[persistence] object AsyncWriteJournal {
+ import JournalProtocol._
+
val successUnit: Success[Unit] = Success(())
+ private val replayIdCounter = new AtomicLong
+
+ private def nextReplayId(): Long = replayIdCounter.incrementAndGet()
+
+ private final case class ReplaySession(
+ replayId: Long,
+ request: ReplayMessages,
+ replyTo: ActorRef,
+ relay: ActorRef,
+ highestSequenceNr: Long,
+ toSequenceNr: Long,
+ nextSequenceNr: Long,
+ remaining: Long,
+ awaitingAck: Boolean)
+
+ private final case class ReplayHighestSequenceNr(replayId: Long,
highestSequenceNr: Long)
+ extends NoSerializationVerificationNeeded
+ private final case class ReplayBatchCompleted(replayId: Long, toSequenceNr:
Long, replayed: Long)
+ extends NoSerializationVerificationNeeded
+ private final case class ReplayFailed(replayId: Long, cause: Throwable)
extends NoSerializationVerificationNeeded
+
+ private def replayForwarderProps(persistentActor: ActorRef, actorInstanceId:
Int): Props =
+ Props(new ReplayForwarder(persistentActor, actorInstanceId))
+
+ private final class ReplayForwarder(persistentActor: ActorRef,
actorInstanceId: Int) extends Actor {
+ def receive = {
+ case response: JournalProtocol.Response =>
+ persistentActor.tell(ReplayBatchResponse(actorInstanceId, response),
Actor.noSender)
+ response match {
+ case _: RecoverySuccess | _: ReplayMessagesFailure =>
context.stop(self)
+ case _ =>
+ }
+ }
+ }
+
final case class Desequenced(msg: Any, snr: Long, target: ActorRef, sender:
ActorRef)
extends NoSerializationVerificationNeeded
diff --git
a/persistence/src/main/scala/org/apache/pekko/persistence/journal/PersistencePluginProxy.scala
b/persistence/src/main/scala/org/apache/pekko/persistence/journal/PersistencePluginProxy.scala
index c06b4b376d..ba02fb9ad4 100644
---
a/persistence/src/main/scala/org/apache/pekko/persistence/journal/PersistencePluginProxy.scala
+++
b/persistence/src/main/scala/org/apache/pekko/persistence/journal/PersistencePluginProxy.scala
@@ -212,7 +212,9 @@ final class PersistencePluginProxy(config: Config) extends
Actor with Stash with
}
case ReplayMessages(_, _, _, _, persistentActor) =>
persistentActor ! ReplayMessagesFailure(timeoutException())
- case DeleteMessagesTo(_, toSequenceNr, persistentActor) =>
+ // The following ReplayMessages request has already failed, so control
messages don't need a response.
+ case ReplayMessagesWithBatching(_) | ReplayMessagesCancel |
ReplayBatchAck(_) | ReplayBatchCancel(_) => ()
+ case DeleteMessagesTo(_, toSequenceNr, persistentActor)
=>
persistentActor ! DeleteMessagesFailure(timeoutException(),
toSequenceNr)
}
diff --git
a/persistence/src/main/scala/org/apache/pekko/persistence/journal/ReplayFilter.scala
b/persistence/src/main/scala/org/apache/pekko/persistence/journal/ReplayFilter.scala
index 02943b7fa6..462908ad73 100644
---
a/persistence/src/main/scala/org/apache/pekko/persistence/journal/ReplayFilter.scala
+++
b/persistence/src/main/scala/org/apache/pekko/persistence/journal/ReplayFilter.scala
@@ -159,6 +159,11 @@ private[pekko] class ReplayFilter(
case e: IllegalStateException if mode == Fail => fail(e)
}
+ case msg: ReplayBatchReady =>
+ if (debugEnabled)
+ log.debug("Replay batch completed: {}", msg)
+ persistentActor.tell(msg, Actor.noSender)
+
case msg @ (_: RecoverySuccess | _: ReplayMessagesFailure) =>
if (debugEnabled)
log.debug("Replay completed: {}", msg)
@@ -183,7 +188,10 @@ private[pekko] class ReplayFilter(
buffer.clear()
persistentActor.tell(ReplayMessagesFailure(cause), Actor.noSender)
context.become {
- case _: ReplayedMessage => // discard
+ case _: ReplayedMessage => // discard
+ case msg: ReplayBatchReady =>
+ sender() ! ReplayBatchCancel(msg.replayId)
+ context.stop(self)
case _: RecoverySuccess | _: ReplayMessagesFailure =>
context.stop(self)
}
diff --git
a/persistence/src/test/scala/org/apache/pekko/persistence/PersistentActorJournalProtocolSpec.scala
b/persistence/src/test/scala/org/apache/pekko/persistence/PersistentActorJournalProtocolSpec.scala
index 711037f6a5..1ae2bdca46 100644
---
a/persistence/src/test/scala/org/apache/pekko/persistence/PersistentActorJournalProtocolSpec.scala
+++
b/persistence/src/test/scala/org/apache/pekko/persistence/PersistentActorJournalProtocolSpec.scala
@@ -144,11 +144,16 @@ class PersistentActorJournalProtocolSpec extends
PekkoSpec(config) with Implicit
}
}
+ def expectReplayMessages(): ReplayMessages = {
+ journal.expectMsgType[ReplayMessagesWithBatching]
+ journal.expectMsgType[ReplayMessages]
+ }
+
def startActor(name: String): ActorRef = {
val subject = system.actorOf(Props(new A(testActor)), name)
subject ! Echo(0)
expectMsg(PreStart(name))
- journal.expectMsgType[ReplayMessages]
+ expectReplayMessages()
journal.reply(RecoverySuccess(0L))
expectMsg(RecoveryCompleted)
expectMsg(Done(0, 0))
@@ -256,7 +261,7 @@ class PersistentActorJournalProtocolSpec extends
PekkoSpec(config) with Implicit
subject ! Fail(new Exception("K-BOOM!"))
expectMsg(PreRestart("test-6"))
expectMsg(PostRestart("test-6"))
- journal.expectMsgType[ReplayMessages]
+ expectReplayMessages()
}
journal.reply(RecoverySuccess(0L))
expectMsg(RecoveryCompleted)
diff --git
a/persistence/src/test/scala/org/apache/pekko/persistence/PersistentActorReplayBatchingSpec.scala
b/persistence/src/test/scala/org/apache/pekko/persistence/PersistentActorReplayBatchingSpec.scala
new file mode 100644
index 0000000000..e0161d9100
--- /dev/null
+++
b/persistence/src/test/scala/org/apache/pekko/persistence/PersistentActorReplayBatchingSpec.scala
@@ -0,0 +1,387 @@
+/*
+ * 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
+
+import scala.collection.immutable
+import scala.concurrent.{ Await, Future, Promise }
+import scala.concurrent.duration._
+import scala.util.Try
+
+import org.apache.pekko.actor.{ ActorIdentity, ActorRef, Identify, Props }
+import org.apache.pekko.persistence.journal.{ AsyncWriteJournal,
SteppingInmemJournal }
+import org.apache.pekko.testkit.{ PekkoSpec, TestProbe }
+
+import com.typesafe.config.ConfigFactory
+
+object RestartingReplayJournal {
+ case object Crash
+ final class CrashException extends RuntimeException("restart the test
journal")
+
+ final case class Control(incarnations: Vector[Future[Unit]], reads:
Vector[Future[Unit]])
+
+ private var incarnationStarted = Vector.empty[Promise[Unit]]
+ private var readStarted = Vector.empty[Promise[Unit]]
+ private var readResults = Vector.empty[Promise[Long]]
+ private var incarnation = 0
+ private var read = 0
+
+ def prepare(): Control = synchronized {
+ incarnationStarted = Vector.fill(2)(Promise[Unit]())
+ readStarted = Vector.fill(2)(Promise[Unit]())
+ readResults = Vector.fill(2)(Promise[Long]())
+ incarnation = 0
+ read = 0
+ Control(incarnationStarted.map(_.future), readStarted.map(_.future))
+ }
+
+ private def onStart(): Unit = synchronized {
+ incarnationStarted(incarnation).success(())
+ incarnation += 1
+ }
+
+ private def nextRead(): Future[Long] = synchronized {
+ val current = read
+ read += 1
+ readStarted(current).success(())
+ readResults(current).future
+ }
+
+ def completeRead(read: Int, highestSequenceNr: Long): Unit = synchronized {
+ readResults(read).success(highestSequenceNr)
+ }
+}
+
+final class RestartingReplayJournal extends AsyncWriteJournal {
+ import RestartingReplayJournal._
+
+ override def preStart(): Unit = {
+ super.preStart()
+ onStart()
+ }
+
+ override def receivePluginInternal: Receive = {
+ case Crash => throw new CrashException
+ }
+
+ override def asyncWriteMessages(
+ messages: immutable.Seq[AtomicWrite]): Future[immutable.Seq[Try[Unit]]]
= Future.successful(Nil)
+
+ override def asyncDeleteMessagesTo(persistenceId: String, toSequenceNr:
Long): Future[Unit] =
+ Future.successful(())
+
+ override def asyncReadHighestSequenceNr(persistenceId: String,
fromSequenceNr: Long): Future[Long] =
+ nextRead()
+
+ override def asyncReplayMessages(persistenceId: String, fromSequenceNr:
Long, toSequenceNr: Long, max: Long)(
+ recoveryCallback: PersistentRepr => Unit): Future[Unit] =
Future.successful(())
+}
+
+object PersistentActorReplayBatchingSpec {
+ val JournalId = "persistent-actor-replay-batching-spec"
+ val RestartingJournalId = "restarting-replay-journal"
+
+ final case class PersistAll(events: Vector[String])
+ final case class Persisted(event: String)
+ final case class DeleteTo(toSequenceNr: Long)
+ final case class Deleted(toSequenceNr: Long)
+ final case class Replayed(event: String, eventSender: ActorRef)
+ case object RecoveryFinished
+
+ val config =
+ SteppingInmemJournal
+ .config(JournalId)
+ .withFallback(ConfigFactory.parseString(s"""
+ pekko.persistence.journal.stepping-inmem {
+ replay-batch-size = 2
+ replay-filter.mode = off
+ }
+ $RestartingJournalId = $${pekko.persistence.journal-plugin-fallback}
+ $RestartingJournalId {
+ class = "org.apache.pekko.persistence.RestartingReplayJournal"
+ replay-batch-size = 2
+ replay-filter.mode = off
+ }
+ """))
+ .withFallback(PersistenceSpec.config("stepping-inmem",
"PersistentActorReplayBatchingSpec"))
+ .withFallback(ConfigFactory.defaultReference())
+ .resolve()
+
+ final class TestActor(override val persistenceId: String, replayMax: Long,
probe: ActorRef) extends PersistentActor {
+ override def recovery: Recovery = Recovery(replayMax = replayMax)
+
+ override def receiveRecover: Receive = {
+ case event: String => probe ! Replayed(event, sender())
+ case RecoveryCompleted => probe ! RecoveryFinished
+ }
+
+ override def receiveCommand: Receive = {
+ case PersistAll(events) =>
+ val replyTo = sender()
+ persistAll(events) { event =>
+ replyTo ! Persisted(event)
+ }
+ case DeleteTo(toSequenceNr) => deleteMessages(toSequenceNr)
+ case DeleteMessagesSuccess(toSequenceNr) => probe ! Deleted(toSequenceNr)
+ }
+ }
+}
+
+class PersistentActorReplayBatchingSpec extends
PekkoSpec(PersistentActorReplayBatchingSpec.config) {
+ import PersistentActorReplayBatchingSpec._
+ import JournalProtocol._
+
+ private def nextBatchedResponse(receiver: TestProbe, actorInstanceId: Int):
JournalProtocol.Response = {
+ val wrapped = receiver.expectMsgType[ReplayBatchResponse]
+ wrapped.actorInstanceId shouldBe actorInstanceId
+ receiver.lastSender shouldBe system.deadLetters
+ wrapped.response
+ }
+
+ private def expectReplayed(receiver: TestProbe, actorInstanceId: Int,
payload: String): Unit =
+ nextBatchedResponse(receiver, actorInstanceId) match {
+ case ReplayedMessage(persistent) => persistent.payload shouldBe payload
+ case other => fail(s"Expected ReplayedMessage, got
[$other]")
+ }
+
+ private def expectBatchReady(receiver: TestProbe, actorInstanceId: Int):
ReplayBatchReady =
+ nextBatchedResponse(receiver, actorInstanceId) match {
+ case ready: ReplayBatchReady => ready
+ case other => fail(s"Expected ReplayBatchReady, got
[$other]")
+ }
+
+ "A PersistentActor recovery" should {
+ "acknowledge each bounded replay batch before the journal produces the
next batch" in {
+ val probe = TestProbe()
+ val persistenceId = "bounded-replay"
+ val events = Vector("a", "b", "c", "d", "e")
+
+ val persisting = system.actorOf(Props(classOf[TestActor], persistenceId,
Long.MaxValue, probe.ref))
+ awaitAssert(SteppingInmemJournal.getRef(JournalId), 3.seconds)
+ val journal = SteppingInmemJournal.getRef(JournalId)
+
+ SteppingInmemJournal.step(journal)
+ probe.expectMsg(RecoveryFinished)
+
+ persisting.tell(PersistAll(events), probe.ref)
+ SteppingInmemJournal.step(journal)
+ events.foreach(event => probe.expectMsg(Persisted(event)))
+
+ watch(persisting)
+ system.stop(persisting)
+ expectTerminated(persisting)
+
+ system.actorOf(Props(classOf[TestActor], persistenceId, Long.MaxValue,
probe.ref))
+ SteppingInmemJournal.step(journal) // read highest sequence number
+
+ SteppingInmemJournal.step(journal) // replay 1-2
+ probe.expectMsg(Replayed("a", system.deadLetters))
+ probe.expectMsg(Replayed("b", system.deadLetters))
+ probe.expectNoMessage(100.millis)
+
+ SteppingInmemJournal.step(journal) // replay 3-4
+ probe.expectMsg(Replayed("c", system.deadLetters))
+ probe.expectMsg(Replayed("d", system.deadLetters))
+ probe.expectNoMessage(100.millis)
+
+ SteppingInmemJournal.step(journal) // replay 5 and complete
+ probe.expectMsg(Replayed("e", system.deadLetters))
+ probe.expectMsg(RecoveryFinished)
+ }
+
+ "preserve the total replayMax limit across batches" in {
+ val probe = TestProbe()
+ val persistenceId = "bounded-replay-max"
+ val events = Vector("a", "b", "c", "d", "e")
+
+ val persisting = system.actorOf(Props(classOf[TestActor], persistenceId,
Long.MaxValue, probe.ref))
+ val journal = SteppingInmemJournal.getRef(JournalId)
+ SteppingInmemJournal.step(journal)
+ probe.expectMsg(RecoveryFinished)
+
+ persisting.tell(PersistAll(events), probe.ref)
+ SteppingInmemJournal.step(journal)
+ events.foreach(event => probe.expectMsg(Persisted(event)))
+ watch(persisting)
+ system.stop(persisting)
+ expectTerminated(persisting)
+
+ system.actorOf(Props(classOf[TestActor], persistenceId, 3L, probe.ref))
+ SteppingInmemJournal.step(journal) // read highest sequence number
+
+ SteppingInmemJournal.step(journal) // replay 1-2
+ probe.expectMsg(Replayed("a", system.deadLetters))
+ probe.expectMsg(Replayed("b", system.deadLetters))
+ probe.expectNoMessage(100.millis)
+
+ SteppingInmemJournal.step(journal) // replay only 3 because replayMax is
now exhausted
+ probe.expectMsg(Replayed("c", system.deadLetters))
+ probe.expectMsg(RecoveryFinished)
+ probe.expectNoMessage(100.millis)
+ }
+
+ "advance across empty sequence number ranges caused by deleted events" in {
+ val probe = TestProbe()
+ val persistenceId = "bounded-replay-deleted-events"
+ val events = Vector("a", "b", "c", "d", "e")
+
+ val persisting = system.actorOf(Props(classOf[TestActor], persistenceId,
Long.MaxValue, probe.ref))
+ val journal = SteppingInmemJournal.getRef(JournalId)
+ SteppingInmemJournal.step(journal)
+ probe.expectMsg(RecoveryFinished)
+
+ persisting.tell(PersistAll(events), probe.ref)
+ SteppingInmemJournal.step(journal)
+ events.foreach(event => probe.expectMsg(Persisted(event)))
+
+ persisting ! DeleteTo(3L)
+ SteppingInmemJournal.step(journal)
+ probe.expectMsg(Deleted(3L))
+ watch(persisting)
+ system.stop(persisting)
+ expectTerminated(persisting)
+
+ system.actorOf(Props(classOf[TestActor], persistenceId, Long.MaxValue,
probe.ref))
+ SteppingInmemJournal.step(journal) // read highest sequence number
+
+ SteppingInmemJournal.step(journal) // replay empty range 1-2
+ probe.expectNoMessage(100.millis)
+
+ SteppingInmemJournal.step(journal) // replay 3-4, with only 4 remaining
+ probe.expectMsg(Replayed("d", system.deadLetters))
+ probe.expectNoMessage(100.millis)
+
+ SteppingInmemJournal.step(journal) // replay 5 and complete
+ probe.expectMsg(Replayed("e", system.deadLetters))
+ probe.expectMsg(RecoveryFinished)
+ }
+
+ "prevent the journal from producing another batch before acknowledgement"
in {
+ val probe = TestProbe()
+ val persistenceId = "withheld-batch-ack"
+ val events = Vector("a", "b", "c", "d", "e")
+
+ val persisting = system.actorOf(Props(classOf[TestActor], persistenceId,
Long.MaxValue, probe.ref))
+ val journal = SteppingInmemJournal.getRef(JournalId)
+ SteppingInmemJournal.step(journal)
+ probe.expectMsg(RecoveryFinished)
+ persisting.tell(PersistAll(events), probe.ref)
+ SteppingInmemJournal.step(journal)
+ events.foreach(event => probe.expectMsg(Persisted(event)))
+ watch(persisting)
+ system.stop(persisting)
+ expectTerminated(persisting)
+
+ val receiver = TestProbe()
+ val actorInstanceId = 17
+ journal.tell(ReplayMessagesWithBatching(actorInstanceId), receiver.ref)
+ journal.tell(ReplayMessages(1L, Long.MaxValue, Long.MaxValue,
persistenceId, receiver.ref), receiver.ref)
+ SteppingInmemJournal.step(journal) // read highest sequence number
+ SteppingInmemJournal.step(journal) // replay 1-2
+
+ expectReplayed(receiver, actorInstanceId, "a")
+ expectReplayed(receiver, actorInstanceId, "b")
+ val firstBatch = expectBatchReady(receiver, actorInstanceId)
+
+ // Make a replay token available, but withhold the acknowledgement. No
second replay call may start.
+ val tokenProbe = TestProbe()
+ journal.tell(SteppingInmemJournal.Token, tokenProbe.ref)
+ receiver.expectNoMessage(100.millis)
+ tokenProbe.expectNoMessage(100.millis)
+
+ journal.tell(ReplayBatchAck(firstBatch.replayId), receiver.ref)
+ tokenProbe.expectMsg(SteppingInmemJournal.TokenConsumed)
+ expectReplayed(receiver, actorInstanceId, "c")
+ expectReplayed(receiver, actorInstanceId, "d")
+ val secondBatch = expectBatchReady(receiver, actorInstanceId)
+
+ journal.tell(ReplayBatchAck(secondBatch.replayId), receiver.ref)
+ SteppingInmemJournal.step(journal)
+ expectReplayed(receiver, actorInstanceId, "e")
+ nextBatchedResponse(receiver, actorInstanceId) shouldBe
RecoverySuccess(5L)
+ }
+
+ "discard an unpaired batching request when its requester terminates" in {
+ val probe = TestProbe()
+ val persistenceId = "terminated-batching-requester"
+ val journal = SteppingInmemJournal.getRef(JournalId)
+
+ val persisting = system.actorOf(Props(classOf[TestActor], persistenceId,
Long.MaxValue, probe.ref))
+ SteppingInmemJournal.step(journal)
+ probe.expectMsg(RecoveryFinished)
+ persisting.tell(PersistAll(Vector("a")), probe.ref)
+ SteppingInmemJournal.step(journal)
+ probe.expectMsg(Persisted("a"))
+ watch(persisting)
+ system.stop(persisting)
+ expectTerminated(persisting)
+
+ val shortLivedRequester = TestProbe()
+ journal.tell(ReplayMessagesWithBatching(actorInstanceId = 23),
shortLivedRequester.ref)
+ journal.tell(Identify("batching-request-registered"), testActor)
+ expectMsg(ActorIdentity("batching-request-registered", Some(journal)))
+ watch(shortLivedRequester.ref)
+ system.stop(shortLivedRequester.ref)
+ expectTerminated(shortLivedRequester.ref)
+ journal.tell(Identify("batching-request-removed"), testActor)
+ expectMsg(ActorIdentity("batching-request-removed", Some(journal)))
+
+ val published = TestProbe()
+ system.eventStream.subscribe(published.ref, classOf[ReplayMessages])
+ journal.tell(
+ ReplayMessages(1L, Long.MaxValue, Long.MaxValue, persistenceId,
shortLivedRequester.ref),
+ shortLivedRequester.ref)
+ published.expectNoMessage(100.millis)
+ SteppingInmemJournal.step(journal)
+ SteppingInmemJournal.step(journal)
+ published.expectMsgType[ReplayMessages].persistenceId shouldBe
persistenceId
+ system.eventStream.unsubscribe(published.ref)
+ }
+
+ "ignore a replay completion from an earlier journal actor incarnation" in {
+ val control = RestartingReplayJournal.prepare()
+ val journal = Persistence(system).journalFor(RestartingJournalId)
+ Await.result(control.incarnations.head, 3.seconds)
+
+ val firstRequester = TestProbe()
+ journal.tell(ReplayMessagesWithBatching(actorInstanceId = 31),
firstRequester.ref)
+ journal.tell(
+ ReplayMessages(1L, Long.MaxValue, Long.MaxValue,
"old-journal-incarnation", firstRequester.ref),
+ firstRequester.ref)
+ Await.result(control.reads.head, 3.seconds)
+
+ journal ! RestartingReplayJournal.Crash
+ Await.result(control.incarnations(1), 3.seconds)
+
+ val secondRequester = TestProbe()
+ journal.tell(ReplayMessagesWithBatching(actorInstanceId = 32),
secondRequester.ref)
+ journal.tell(
+ ReplayMessages(1L, Long.MaxValue, Long.MaxValue,
"new-journal-incarnation", secondRequester.ref),
+ secondRequester.ref)
+ Await.result(control.reads(1), 3.seconds)
+
+ RestartingReplayJournal.completeRead(read = 0, highestSequenceNr = 0L)
+ secondRequester.expectNoMessage(100.millis)
+
+ RestartingReplayJournal.completeRead(read = 1, highestSequenceNr = 0L)
+ val response = secondRequester.expectMsgType[ReplayBatchResponse]
+ response.actorInstanceId shouldBe 32
+ response.response shouldBe RecoverySuccess(0L)
+ secondRequester.lastSender shouldBe system.deadLetters
+ }
+ }
+}
diff --git
a/persistence/src/test/scala/org/apache/pekko/persistence/journal/ReplayFilterSpec.scala
b/persistence/src/test/scala/org/apache/pekko/persistence/journal/ReplayFilterSpec.scala
index dd792984f1..ab4890770a 100644
---
a/persistence/src/test/scala/org/apache/pekko/persistence/journal/ReplayFilterSpec.scala
+++
b/persistence/src/test/scala/org/apache/pekko/persistence/journal/ReplayFilterSpec.scala
@@ -157,6 +157,33 @@ class ReplayFilterSpec extends PekkoSpec with
ImplicitSender {
}
"ReplayFilter in Fail mode" must {
+ "keep its buffer and writer state across bounded replay batches" in {
+ val journalProbe = TestProbe()
+ val filter = system.actorOf(
+ ReplayFilter.props(testActor, mode = Fail, windowSize = 100,
maxOldWriters = 10, debugEnabled = false))
+ val m2b = m2.copy(persistent = m2.persistent.update(writerUuid =
writerB))
+ val batchReady = ReplayBatchReady(1L)
+
+ filter.tell(m1, journalProbe.ref)
+ filter.tell(m2b, journalProbe.ref)
+ filter.tell(batchReady, journalProbe.ref)
+
+ // The marker passes through without flushing the look-ahead buffer.
+ expectMsg(batchReady)
+
+ EventFilter.error(start = "Invalid replayed event", occurrences =
1).intercept {
+ filter.tell(m2, journalProbe.ref) // writerA is now an old writer
+ expectMsgType[ReplayMessagesFailure].cause.getClass should
be(classOf[IllegalStateException])
+
+ // The failed filter cancels the replay session when the journal
finishes the current batch.
+ filter.tell(batchReady, journalProbe.ref)
+ journalProbe.expectMsg(ReplayBatchCancel(1L))
+ }
+
+ watch(filter)
+ expectTerminated(filter)
+ }
+
"fail when message with same seqNo from old overlapping writer" in {
val filter = system.actorOf(
ReplayFilter.props(testActor, mode = Fail, windowSize = 100,
maxOldWriters = 10, debugEnabled = false))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
