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
Akka persistence - remove messages from journal (or mark as confirmed), when recieve confirmation of message delivery Akka	1	732	December 19, 2019
My Bug while Using Multiple Cluster sharding with Akka Persistence Persistence / Event Sourcing akka-typed , java , akka-cluster	0	657	November 5, 2020
Is there a way to disable replay journal when recovery? Persistence / Event Sourcing	3	1229	March 22, 2019
Akka sharding and persistence recovery Akka	5	649	August 20, 2019
Akka-Cluster: "Failed to persist event" => "Retry request for shard" Akka akka-cluster	0	478	October 28, 2020

JournalFailureException not propagated to the Terminate signal

Related Topics