From 4fd567fdf32cb275710788faa41b27722552564d Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E8=99=8E=E9=B8=A3?= <hepin.p@alibaba-inc.com>
Date: Sat, 11 Jul 2026 18:09:02 +0800
Subject: [PATCH] fix: bound persistence event replay batches

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 +-
 ...entSourcedBehaviorReplayBatchingSpec.scala | 243 +++++++++++
 .../typed/internal/ExternalInteractions.scala |   3 +
 .../typed/internal/ReplayingEvents.scala      |  10 +
 persistence/src/main/resources/reference.conf |   5 +
 .../pekko/persistence/Eventsourced.scala      |  13 +-
 .../pekko/persistence/JournalProtocol.scala   |  35 ++
 .../journal/AsyncWriteJournal.scala           | 316 ++++++++++++--
 .../journal/PersistencePluginProxy.scala      |   4 +-
 .../persistence/journal/ReplayFilter.scala    |  10 +-
 .../PersistentActorJournalProtocolSpec.scala  |   9 +-
 .../PersistentActorReplayBatchingSpec.scala   | 387 ++++++++++++++++++
 .../journal/ReplayFilterSpec.scala            |  27 ++
 15 files changed, 1043 insertions(+), 55 deletions(-)
 create mode 100644 persistence-typed-tests/src/test/scala/org/apache/pekko/persistence/typed/scaladsl/EventSourcedBehaviorReplayBatchingSpec.scala
 create mode 100644 persistence/src/test/scala/org/apache/pekko/persistence/PersistentActorReplayBatchingSpec.scala

diff --git a/docs/src/main/paradox/persistence.md b/docs/src/main/paradox/persistence.md
index 900f6d27950..36d0f92514c 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 3e4c2aea658..172067f0bb9 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 3364cfef336..01baf79c63e 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 00000000000..ba8b0726d0e
--- /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 10f8b6e4977..134fe4c4eba 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 db902d9bb6b..50abf443479 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 ff34a035bb1..c00b1f5763b 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 cb2006526cd..e0593256ab4 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 d82b4279062..c23ecfc4693 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 ae27bd5c4fe..6afe37527e2 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 c06b4b376d8..ba02fb9ad46 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 02943b7fa6e..462908ad73f 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 711037f6a50..1ae2bdca46d 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 00000000000..e0161d91004
--- /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 dd792984f1b..ab4890770ac 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))