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 1fa250d450 Stop PersistentActor by default on RecoveryCompleted
failure (#3359)
1fa250d450 is described below
commit 1fa250d450e18bbebd0374378b6e2a8069473e46
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Fri Jul 17 19:56:29 2026 +0800
Stop PersistentActor by default on RecoveryCompleted failure (#3359)
Motivation:
A classic PersistentActor can restart indefinitely under the default
supervision strategy when its RecoveryCompleted handler throws.
Modification:
Wrap exceptions from the RecoveryCompleted handler in
ActorInitializationException before passing them to the parent supervisor. Keep
recovery permit cleanup in the existing outer catch, add directional tests for
the default Stop decision and a custom parent Restart decision, and document
the supervision contract.
Result:
The default strategy stops the actor instead of looping, while a custom
parent supervisor can still resume or restart it. The original failure remains
available as the exception cause.
Tests:
- sbt "persistence / Test / testOnly
org.apache.pekko.persistence.RecoveryPermitterSpec" (7 passed; restart test
failed before the fix)
- sbt "persistence / Test / test" (62 passed)
- sbt docs/paradox
- sbt headerCreateAll
- sbt +headerCheckAll
- sbt checkCodeStyle
- sbt validatePullRequest (24m44s; failed only on unchanged
baseline/environment issues: x86 LevelDB JNI on macOS ARM64 and
IntegrationDocTest using pekko.dispatch.UnboundedMailbox)
- sbt sortImports (not completed: repository Scalafix rule failed with
scala.meta NoSuchMethodError)
- scalafmt --list --mode diff-ref=origin/main
- git diff --check
- Qoder stdout review: No must-fix findings
References:
Fixes #3234, Fixes #3249
---
docs/src/main/paradox/persistence.md | 4 ++
.../apache/pekko/persistence/Eventsourced.scala | 14 ++++--
.../pekko/persistence/RecoveryPermitterSpec.scala | 57 ++++++++++++++++++----
3 files changed, 62 insertions(+), 13 deletions(-)
diff --git a/docs/src/main/paradox/persistence.md
b/docs/src/main/paradox/persistence.md
index 36d0f92514..2c0c696760 100644
--- a/docs/src/main/paradox/persistence.md
+++ b/docs/src/main/paradox/persistence.md
@@ -215,6 +215,10 @@ The actor will always receive a `RecoveryCompleted`
message, even if there are n
in the journal and the snapshot store is empty, or if it's a new persistent
actor with a previously
unused `persistenceId`.
+If handling `RecoveryCompleted` throws an exception, the failure is reported
to the parent supervisor
+as an `ActorInitializationException`. The default supervision strategy stops
the actor, while a custom
+supervision strategy may resume or restart it.
+
If there is a problem with recovering the state of the actor from the journal,
`onRecoveryFailure`
is called (logging the error by default) and the actor will be stopped.
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 e0593256ab..ccdb90c9eb 100644
--- a/persistence/src/main/scala/org/apache/pekko/persistence/Eventsourced.scala
+++ b/persistence/src/main/scala/org/apache/pekko/persistence/Eventsourced.scala
@@ -22,7 +22,7 @@ import scala.concurrent.duration.FiniteDuration
import scala.util.control.NonFatal
import org.apache.pekko
-import pekko.actor.{ Actor, ActorCell, DeadLetter, StashOverflowException }
+import pekko.actor.{ Actor, ActorCell, ActorInitializationException,
DeadLetter, StashOverflowException }
import pekko.annotation.{ InternalApi, InternalStableApi }
import pekko.dispatch.Envelope
import pekko.event.{ Logging, LoggingAdapter }
@@ -792,9 +792,15 @@ private[persistence] trait Eventsourced
sequenceNr = highestSeqNr
setLastSequenceNr(highestSeqNr)
_recoveryRunning = false
- try Eventsourced.super.aroundReceive(recoveryBehavior,
RecoveryCompleted)
- finally transitToProcessingState() // in finally in case
exception and resume strategy
- // if exception from RecoveryCompleted the permit is returned in
below catch
+ try {
+ Eventsourced.super.aroundReceive(recoveryBehavior,
RecoveryCompleted)
+ } catch {
+ case NonFatal(t) =>
+ throw ActorInitializationException(
+ self,
+ s"Exception in receiveRecover when handling
RecoveryCompleted for persistenceId [$persistenceId]",
+ t)
+ } finally transitToProcessingState()
returnRecoveryPermit()
case ReplayMessagesFailure(cause) =>
timeoutCancellable.cancel()
diff --git
a/persistence/src/test/scala/org/apache/pekko/persistence/RecoveryPermitterSpec.scala
b/persistence/src/test/scala/org/apache/pekko/persistence/RecoveryPermitterSpec.scala
index bf4a91f98b..46af4842a2 100644
---
a/persistence/src/test/scala/org/apache/pekko/persistence/RecoveryPermitterSpec.scala
+++
b/persistence/src/test/scala/org/apache/pekko/persistence/RecoveryPermitterSpec.scala
@@ -27,6 +27,7 @@ import com.typesafe.config.ConfigFactory
object RecoveryPermitterSpec {
class TestExc extends RuntimeException("simulated exc") with NoStackTrace
+ final case class Supervised(cause: Throwable)
def testProps(name: String, probe: ActorRef, throwFromRecoveryCompleted:
Boolean = false): Props =
Props(new TestPersistentActor(name, probe, throwFromRecoveryCompleted))
@@ -46,12 +47,30 @@ object RecoveryPermitterSpec {
if (throwFromRecoveryCompleted)
throw new TestExc
}
+
override def receiveCommand: Receive = {
case "stop" =>
context.stop(self)
}
}
+ class RestartOnceSupervisor(childProps: Props, probe: ActorRef) extends
Actor {
+ private var failures = 0
+
+ override val supervisorStrategy = OneForOneStrategy(loggingEnabled =
false) {
+ case cause: ActorInitializationException =>
+ failures += 1
+ probe ! Supervised(cause)
+ if (failures == 1) SupervisorStrategy.Restart
+ else SupervisorStrategy.Stop
+ }
+
+ override def preStart(): Unit =
+ probe ! context.actorOf(childProps)
+
+ override def receive: Receive = Actor.emptyBehavior
+ }
+
}
class RecoveryPermitterSpec extends
PersistenceSpec(ConfigFactory.parseString(s"""
@@ -62,7 +81,7 @@ class RecoveryPermitterSpec extends
PersistenceSpec(ConfigFactory.parseString(s"
import RecoveryPermitter._
import RecoveryPermitterSpec._
- system.eventStream.publish(TestEvent.Mute(EventFilter[TestExc]()))
+
system.eventStream.publish(TestEvent.Mute(EventFilter[ActorInitializationException]()))
val permitter = Persistence(system).recoveryPermitter
val p1 = TestProbe()
@@ -171,23 +190,43 @@ class RecoveryPermitterSpec extends
PersistenceSpec(ConfigFactory.parseString(s"
permitter.tell(ReturnRecoveryPermit, p4.ref)
}
- "return permit when actor throws from RecoveryCompleted" in {
+ "return permit and stop actor by default when it throws from
RecoveryCompleted" in {
requestPermit(p1)
requestPermit(p2)
val persistentActor = system.actorOf(testProps("p3", p3.ref,
throwFromRecoveryCompleted = true))
+ p3.watch(persistentActor)
p3.expectMsg(RecoveryCompleted)
p3.expectMsg("postStop")
- // it's restarting
- (1 to 5).foreach { _ =>
+ p3.expectTerminated(persistentActor)
+
+ requestPermit(p4)
+
+ permitter.tell(ReturnRecoveryPermit, p1.ref)
+ permitter.tell(ReturnRecoveryPermit, p2.ref)
+ permitter.tell(ReturnRecoveryPermit, p4.ref)
+ }
+
+ "allow the parent supervisor to restart an actor that throws from
RecoveryCompleted" in {
+ requestPermit(p1)
+ requestPermit(p2)
+
+ val supervisor = system.actorOf(
+ Props(new RestartOnceSupervisor(testProps("p3", p3.ref,
throwFromRecoveryCompleted = true), p3.ref)))
+ val persistentActor = p3.expectMsgType[ActorRef]
+ p3.watch(persistentActor)
+
+ (1 to 2).foreach { _ =>
p3.expectMsg(RecoveryCompleted)
+ p3.lastSender shouldBe persistentActor
+ p3.expectMsgPF(hint = "supervised ActorInitializationException caused
by TestExc") {
+ case Supervised(cause: ActorInitializationException)
+ if cause.getActor == persistentActor &&
cause.getCause.isInstanceOf[TestExc] =>
+ }
p3.expectMsg("postStop")
}
- // stop it
- val stopProbe = TestProbe()
- stopProbe.watch(persistentActor)
- system.stop(persistentActor)
- stopProbe.expectTerminated(persistentActor)
+ p3.expectTerminated(persistentActor)
+ system.stop(supervisor)
requestPermit(p4)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
