Same logic for onRecoveryFailure in ReplayingSnapshot as ReplayingEvents, #27526 (#27530)

This commit is contained in:
Patrik Nordwall 2019-10-01 12:24:36 +02:00 committed by Arnout Engelen
parent f6ebff6e27
commit b10a83ad9b
3 changed files with 18 additions and 7 deletions

View file

@ -186,9 +186,10 @@ private[akka] final class ReplayingEvents[C, E, S](
} }
/** /**
* Called whenever a message replay fails. By default it logs the error. * Called whenever a message replay fails.
* *
* The actor is always stopped after this method has been invoked. * This method throws `JournalFailureException` which will be caught by the internal
* supervision strategy to stop or restart the actor with backoff.
* *
* @param cause failure cause. * @param cause failure cause.
* @param event the event that was being processed when the exception was thrown * @param event the event that was being processed when the exception was thrown

View file

@ -11,6 +11,7 @@ import akka.annotation.{ InternalApi, InternalStableApi }
import akka.persistence.SnapshotProtocol.LoadSnapshotFailed import akka.persistence.SnapshotProtocol.LoadSnapshotFailed
import akka.persistence.SnapshotProtocol.LoadSnapshotResult import akka.persistence.SnapshotProtocol.LoadSnapshotResult
import akka.persistence._ import akka.persistence._
import akka.persistence.typed.RecoveryFailed
import akka.util.unused import akka.util.unused
/** /**
@ -77,17 +78,26 @@ private[akka] class ReplayingSnapshot[C, E, S](override val setup: BehaviorSetup
} }
/** /**
* Called whenever a message replay fails. By default it logs the error. * Called whenever snapshot recovery fails.
* *
* The actor is always stopped after this method has been invoked. * This method throws `JournalFailureException` which will be caught by the internal
* supervision strategy to stop or restart the actor with backoff.
* *
* @param cause failure cause. * @param cause failure cause.
*/ */
private def onRecoveryFailure(cause: Throwable): Behavior[InternalProtocol] = { private def onRecoveryFailure(cause: Throwable): Behavior[InternalProtocol] = {
onRecoveryFailed(setup.context, cause) onRecoveryFailed(setup.context, cause)
setup.onSignal(setup.emptyState, RecoveryFailed(cause), catchAndLog = true)
setup.cancelRecoveryTimer() setup.cancelRecoveryTimer()
setup.log.error(s"Persistence failure when replaying snapshot, due to: ${cause.getMessage}", cause)
Behaviors.stopped tryReturnRecoveryPermit("on snapshot recovery failure: " + cause.getMessage)
if (setup.log.isDebugEnabled)
setup.log.debug("Recovery failure for persistenceId [{}]", setup.persistenceId)
val msg = s"Exception during recovery from snapshot. " +
s"PersistenceId [${setup.persistenceId.id}]. ${cause.getMessage}"
throw new JournalFailureException(msg, cause)
} }
@InternalStableApi @InternalStableApi

View file

@ -496,7 +496,7 @@ class EventSourcedBehaviorSpec
} }
"fail after recovery timeout" in { "fail after recovery timeout" in {
LoggingEventFilter.error("Persistence failure when replaying snapshot").intercept { LoggingEventFilter.error("Exception during recovery from snapshot").intercept {
val c = spawn( val c = spawn(
Behaviors.setup[Command](ctx => Behaviors.setup[Command](ctx =>
counter(ctx, nextPid) counter(ctx, nextPid)