JournalFailureException not propagated to the Terminate signal

vigoo · November 8, 2018, 12:37pm

When a persistent, typed actor fails during recovery (for example because a breaking change in the serialization format), it throws a JournalFailureException.
I’m watching this actor from another one and get the Terminated signal but its failure is set to None so there is no way to determine if the actor stopped because of the journal failure or in a normal way.

The exception is thrown here:

github.com

akka/akka/blob/784e300343efa16df07b361255b6db91ebf38f36/akka-persistence-typed/src/main/scala/akka/persistence/typed/internal/EventsourcedReplayingEvents.scala#L143-L164


/**
 * Called whenever a message replay fails. By default it logs the error.
 *
 * The actor is always stopped after this method has been invoked.
 *
 * @param cause failure cause.
 * @param message the message that was being processed when the exception was thrown
 */
protected def onRecoveryFailure(cause: Throwable, sequenceNr: Long, message: Option[Any]): Behavior[InternalProtocol] = {
  setup.cancelRecoveryTimer()
  tryReturnRecoveryPermit("on replay failure: " + cause.getMessage)


  val msg = message match {
    case Some(evt) ⇒
      s"Exception during recovery while handling [${evt.getClass.getName}] with sequence number [$sequenceNr]. " +
        s"PersistenceId [${setup.persistenceId.id}]"
    case None ⇒
      s"Exception during recovery.  Last known sequence number [$sequenceNr]. PersistenceId [${setup.persistenceId.id}]"
  }

This file has been truncated. show original

which is then captured by a supervisor behavior here:

github.com

akka/akka/blob/784e300343efa16df07b361255b6db91ebf38f36/akka-persistence-typed/src/main/scala/akka/persistence/typed/internal/PersistentBehaviorImpl.scala#L107


      }
      val widened = EventsourcedRequestingRecoveryPermit(eventsourcedSetup).widen[Any] {
        case res: JournalProtocol.Response           ⇒ InternalProtocol.JournalResponse(res)
        case res: SnapshotProtocol.Response          ⇒ InternalProtocol.SnapshotterResponse(res)
        case RecoveryPermitter.RecoveryPermitGranted ⇒ InternalProtocol.RecoveryPermitGranted
        case internal: InternalProtocol              ⇒ internal // such as RecoveryTickEvent
        case cmd: Command @unchecked                 ⇒ InternalProtocol.IncomingCommand(cmd)
      }
      Behaviors.intercept(onStopInterceptor)(widened).narrow[Command]
    }
  }.onFailure[JournalFailureException](supervisionStrategy)
}


/**
 * The `callback` function is called to notify the actor that the recovery process
 * is finished.
 */
def onRecoveryCompleted(callback: State ⇒ Unit): PersistentBehavior[Command, Event, State] =
  copy(recoveryCompleted = callback)


/**

which logs the exception and stops the actor.

Is this behavior intentional? I could not find a way to customize the behavior so it propagates the failure to the watchers.

lay · January 29, 2021, 9:05am

I just found this more than 2 years old question while I am struggeling with the same problem. I just converted a classic event source actor to the new typed API. I found no way to report a Journal write failure to the parent actor.

While I understand the reasoning behind the Akka implementation - the user shall not be able to resume an EventSourcedBehavior on a JournalFailureException - it would good to have a kind of “hook” to somehow notify the failure to other actors that might be interested in it.

Wouldn’t be a signal WriteJournalFailed a good solution?

Topic		Replies	Views
Exception during recovery. Last known sequence number [0]. PersistenceId [abc], due to: null Persistence / Event Sourcing	3	2404	June 5, 2020
How to persist messages with new sequence Number in case of Recovery Failures Persistence / Event Sourcing akka , java	0	468	October 26, 2020
Corrupted Event Journal in Akka Persistence Akka akka-typed	3	218	May 10, 2024
Akka Streams Resuming Materializer does not log exceptions Akka Streams & Alpakka	2	1327	December 16, 2021
How to handle exception during creation? Akka	1	2771	September 10, 2019

JournalFailureException not propagated to the Terminate signal

Related Topics