Persistence

Dependency

To use Akka Persistence Typed, add the module to your project:

sbt
libraryDependencies += "com.typesafe.akka" %% "akka-persistence-typed" % "2.5.16"
Maven
<dependency>
  <groupId>com.typesafe.akka</groupId>
  <artifactId>akka-persistence-typed_2.11</artifactId>
  <version>2.5.16</version>
</dependency>
Gradle
dependencies {
  compile group: 'com.typesafe.akka', name: 'akka-persistence-typed_2.11', version: '2.5.16'
}

Introduction

Akka Persistence is a library for building event sourced actors. For background about how it works see the untyped Akka Persistence section. This documentation shows how the typed API for persistence works and assumes you know what is meant by Command, Event and State.

Warning

This module is currently marked as may change in the sense of being the subject of active research. This means that API or semantics can change without warning or deprecation period and it is not recommended to use this module in production just yet—you have been warned.

Example

Let’s start with a simple example. The minimum required for a PersistentBehavior is:

Scala
sealed trait Command
sealed trait Event
case class State()

val behavior: Behavior[Command] =
  PersistentBehaviors.receive[Command, Event, State](
    persistenceId = "abc",
    emptyState = State(),
    commandHandler =
      (ctx, state, cmd) ⇒
        throw new RuntimeException("TODO: process the command & return an Effect"),
    eventHandler =
      (state, evt) ⇒
        throw new RuntimeException("TODO: process the event return the next state")
  )
Full source at GitHub
Java
public interface Command {}
public interface Event {}
public static class State {}

public static class MyPersistentBehavior extends PersistentBehavior<Command, Event, State> {
  public MyPersistentBehavior(String persistenceId) {
    super(persistenceId, SupervisorStrategy.restartWithBackoff(Duration.ofSeconds(10), Duration.ofSeconds(30), 0.2));
  }

  @Override
  public State emptyState() {
    return new State();
  }

  @Override
  public CommandHandler<Command, Event, State> commandHandler() {
    return (state, command) -> {
      throw new RuntimeException("TODO: process the command & return an Effect");
    };
  }

  @Override
  public EventHandler<State, Event>  eventHandler() {
    return (state, event) -> {
      throw new RuntimeException("TODO: process the event return the next state");
    };
  }

  @Override
  public void onRecoveryCompleted(State state) {
    throw new RuntimeException("TODO: add some end-of-recovery side-effect here");
  }

  @Override
  public Set<String> tagsFor(Event event) {
    throw new RuntimeException("TODO: inspect the event and return any tags it should have");
  }
}

static PersistentBehavior<Command, Event, State> persistentBehavior = new MyPersistentBehavior("pid");
Full source at GitHub

The first important thing to notice is the Behavior of a persistent actor is typed to the type of the Command because this is the type of message a persistent actor should receive. In Akka Typed this is now enforced by the type system. The event and state are only used internally.

The components that make up a PersistentBehavior are:

  • persistenceId is the stable unique identifier for the persistent actor.
  • emptyState defines the State when the entity is first created e.g. a Counter would start with 0 as state.
  • commandHandler defines how to handle command by producing Effects e.g. persisting events, stopping the persistent actor.
  • eventHandler returns the new state given the current state when an event has been persisted.

Next we’ll discuss each of these in detail.

Command handler

The command handler is a function with 2 parameters for3 parameters for the ActorContext, current State and Command.

A command handler returns an Effect directive that defines what event or events, if any, to persist. Effects are created using a factory that is returned via the Effect() method the Effect factory and can be one of:

  • persist will persist one single event or several events atomically, i.e. all events are stored or none of them are stored if there is an error
  • none no events are to be persisted, for example a read-only command
  • unhandled the command is unhandled (not supported) in current state
  • stop stop this actor

In addition to returning the primary Effect for the command PersistentBehaviors can also chain side effects (SideEffects) are to be performed after successful persist which is achieved with the andThen and thenRun function e.g Effect.persist(..).andThenEffect().persist(..).andThen. The thenRun function is a convenience around creating a SideEffect.

In the example below a reply is sent to the replyTo ActorRef. Note that the new state after applying the event is passed as parameter to the thenRun function. All thenRun registered callbacks are executed sequentially after successful execution of the persist statement (or immediately, in case of none and unhandled).

Event handler

When an event has been persisted successfully the new state is created by applying the event to the current state with the eventHandler.

The event handler returns the new state, which must be immutable so you return a new instance of the state. The same event handler is also used when the entity is started up to recover its state from the stored events.

It is not recommended to perform side effects in the event handler, as those are also executed during recovery of an persistent actor

Basic example

Command and event:

Scala
sealed trait SimpleCommand
case class Cmd(data: String) extends SimpleCommand

sealed trait SimpleEvent
case class Evt(data: String) extends SimpleEvent
Full source at GitHub
Java
public static class SimpleCommand {
  public final String data;

  public SimpleCommand(String data) {
    this.data = data;
  }
}
Full source at GitHub

State is a List containing all the events:

Scala
case class ExampleState(events: List[String] = Nil)
Full source at GitHub
Java
static class SimpleState {
  private final List<String> events;

  SimpleState(List<String> events) {
    this.events = events;
  }

  SimpleState() {
    this.events = new ArrayList<>();
  }


  SimpleState addEvent(SimpleEvent event) {
    List<String> newEvents = new ArrayList<>(events);
    newEvents.add(event.data);
    return new SimpleState(newEvents);
  }
}
Full source at GitHub

The command handler persists the Cmd payload in an Evt. In this simple example the command handler is defined using a lambda, for the more complicated example below a CommandHandlerBuilder is used:

Scala
val commandHandler: CommandHandler[SimpleCommand, SimpleEvent, ExampleState] =
  CommandHandler.command {
    case Cmd(data) ⇒ Effect.persist(Evt(data))
  }
Full source at GitHub
Java
@Override
public CommandHandler<SimpleCommand, SimpleEvent, SimpleState> commandHandler() {
  return (state, cmd) -> Effect().persist(new SimpleEvent(cmd.data));
}
Full source at GitHub

The event handler appends the event to the state. This is called after successfully persisting the event in the database . As with the command handler the event handler is defined using a lambda, see below for a more complicated example using the EventHandlerBuilder:

Scala
val eventHandler: (ExampleState, SimpleEvent) ⇒ ExampleState = {
  case (state, Evt(data)) ⇒ state.copy(data :: state.events)
}
Full source at GitHub
Java
@Override
public EventHandler<SimpleState, SimpleEvent> eventHandler() {
  return (state, event) -> state.addEvent(event);
}
Full source at GitHub

These are used to create a PersistentBehavior:

Scala
val simpleBehavior: PersistentBehavior[SimpleCommand, SimpleEvent, ExampleState] =
  PersistentBehaviors.receive[SimpleCommand, SimpleEvent, ExampleState](
    persistenceId = "sample-id-1",
    emptyState = ExampleState(Nil),
    commandHandler = commandHandler,
    eventHandler = eventHandler)
Full source at GitHub
Java
public static PersistentBehavior<SimpleCommand, SimpleEvent, SimpleState> pb = new PersistentBehavior<SimpleCommand, SimpleEvent, SimpleState>("p1") {
  @Override
  public SimpleState emptyState() {
    return new SimpleState();
  }

  @Override
  public CommandHandler<SimpleCommand, SimpleEvent, SimpleState> commandHandler() {
    return (state, cmd) -> Effect().persist(new SimpleEvent(cmd.data));
  }

  @Override
  public EventHandler<SimpleState, SimpleEvent> eventHandler() {
    return (state, event) -> state.addEvent(event);
  }

  @Override
  public EventAdapter<SimpleEvent, Wrapper<SimpleEvent>> eventAdapter() {
    return new EventAdapterExample();
  }
};
Full source at GitHub

The PersistentBehavior can then be run as with any plain typed actor as described in typed actors documentation.

The ActorContext can be obtained with Behaviors.setup and be passed as a constructor parameter.

Larger example

After processing a message, plain typed actors are able to return the Behavior that is used for next message.

As you can see in the above examples this is not supported by typed persistent actors. Instead, the state is returned by eventHandler. The reason a new behavior can’t be returned is that behavior is part of the actor’s state and must also carefully be reconstructed during recovery. If it would have been supported it would mean that the behavior must be restored when replaying events and also encoded in the state anyway when snapshots are used. That would be very prone to mistakes and thus not allowed in Typed Persistence.

For basic actors you can use the same set of command handlers independent of what state the entity is in, as shown in above example. For more complex actors it’s useful to be able to change the behavior in the sense that different functions for processing commands may be defined depending on what state the actor is in. This is useful when implementing finite state machine (FSM) like entities.

The next example shows how to define different behavior based on the current State. It is an actor that represents the state of a blog post. Before a post is started the only command it can process is to AddPost. Once it is started then it we can look it up with GetPost, modify it with ChangeBody or publish it with Publish.

The state is captured by:

Scala
object BlogState {
  val empty = BlogState(None, published = false)
}

final case class BlogState(content: Option[PostContent], published: Boolean) {
  def withContent(newContent: PostContent): BlogState =
    copy(content = Some(newContent))
  def isEmpty: Boolean = content.isEmpty
  def postId: String = content match {
    case Some(c) ⇒ c.postId
    case None    ⇒ throw new IllegalStateException("postId unknown before post is created")
  }
}
Full source at GitHub
Java
interface BlogState {}

public static class BlankState implements BlogState {}

public static class DraftState implements BlogState {
  final PostContent postContent;
  final boolean published;

  DraftState(PostContent postContent, boolean published) {
    this.postContent = postContent;
    this.published = published;
  }

  public DraftState withContent(PostContent newContent) {
    return new DraftState(newContent, this.published);
  }

  public String postId() {
    return postContent.postId;
  }
}

public static class PublishedState implements BlogState {
  final PostContent postContent;

  PublishedState(PostContent postContent) {
    this.postContent = postContent;
  }

  public PublishedState withContent(PostContent newContent) {
    return new PublishedState(newContent);
  }

  public String postId() {
    return postContent.postId;
  }
}
Full source at GitHub

The commands, of which only a subset are valid depending on the state:

Scala
sealed trait BlogCommand extends Serializable
final case class AddPost(content: PostContent, replyTo: ActorRef[AddPostDone]) extends BlogCommand
final case class AddPostDone(postId: String)
final case class GetPost(replyTo: ActorRef[PostContent]) extends BlogCommand
final case class ChangeBody(newBody: String, replyTo: ActorRef[Done]) extends BlogCommand
final case class Publish(replyTo: ActorRef[Done]) extends BlogCommand
final case object PassivatePost extends BlogCommand
final case class PostContent(postId: String, title: String, body: String)
Full source at GitHub
Java
public interface BlogCommand {
}
public static class AddPost implements BlogCommand {
  final PostContent content;
  final ActorRef<AddPostDone> replyTo;

  public AddPost(PostContent content, ActorRef<AddPostDone> replyTo) {
    this.content = content;
    this.replyTo = replyTo;
  }
}
public static class AddPostDone implements BlogCommand {
  final String postId;

  public AddPostDone(String postId) {
    this.postId = postId;
  }
}
public static class GetPost implements BlogCommand {
  final ActorRef<PostContent> replyTo;

  public GetPost(ActorRef<PostContent> replyTo) {
    this.replyTo = replyTo;
  }
}
public static class ChangeBody implements BlogCommand {
  final String newBody;
  final ActorRef<Done> replyTo;

  public ChangeBody(String newBody, ActorRef<Done> replyTo) {
    this.newBody = newBody;
    this.replyTo = replyTo;
  }
}
public static class Publish implements BlogCommand {
  final ActorRef<Done> replyTo;

  public Publish(ActorRef<Done> replyTo) {
    this.replyTo = replyTo;
  }
}
public static class PassivatePost implements BlogCommand {

}
public static class PostContent implements BlogCommand {
  final String postId;
  final String title;
  final String body;

  public PostContent(String postId, String title, String body) {
    this.postId = postId;
    this.title = title;
    this.body = body;
  }
}
Full source at GitHub

The commandler handler to process each command is decided by the state class (or state predicate) that is given to the commandHandlerBuilder and the match cases in the builders. Several builders can be composed with orElse: The command handler to process each command is decided by a CommandHandler.byState command handler, which is a function from State => CommandHandler:

Scala
private val commandHandler: (ActorContext[BlogCommand], BlogState, BlogCommand) ⇒ Effect[BlogEvent, BlogState] =
  CommandHandler.byState {
    case state if state.isEmpty  ⇒ initial
    case state if !state.isEmpty ⇒ postAdded
  }
Full source at GitHub
Java
@Override
public CommandHandler<BlogCommand, BlogEvent, BlogState> commandHandler() {
  return
      initialCommandHandler()
          .orElse(draftCommandHandler())
          .orElse(publishedCommandHandler())
          .orElse(commonCommandHandler())
          .build();
}
Full source at GitHub

The CommandHandlerBuilderCommandHandler for a post that hasn’t been initialized with content:

Scala
private val initial: (ActorContext[BlogCommand], BlogState, BlogCommand) ⇒ Effect[BlogEvent, BlogState] =
  (ctx, state, cmd) ⇒
    cmd match {
      case AddPost(content, replyTo) ⇒
        val evt = PostAdded(content.postId, content)
        Effect.persist(evt).thenRun { state2 ⇒
          // After persist is done additional side effects can be performed
          replyTo ! AddPostDone(content.postId)
        }
      case PassivatePost ⇒
        Effect.stop
      case _ ⇒
        Effect.unhandled
    }
Full source at GitHub
Java
private CommandHandlerBuilder<BlogCommand, BlogEvent, BlankState, BlogState> initialCommandHandler() {
  return commandHandlerBuilder(BlankState.class)
      .matchCommand(AddPost.class, (state, cmd) -> {
        PostAdded event = new PostAdded(cmd.content.postId, cmd.content);
        return Effect().persist(event)
            .andThen(() -> cmd.replyTo.tell(new AddPostDone(cmd.content.postId)));
      });
}
Full source at GitHub

And a different CommandHandlerBuilderCommandHandler for after the post content has been added:

Scala
private val postAdded: (ActorContext[BlogCommand], BlogState, BlogCommand) ⇒ Effect[BlogEvent, BlogState] = {
  (ctx, state, cmd) ⇒
    cmd match {
      case ChangeBody(newBody, replyTo) ⇒
        val evt = BodyChanged(state.postId, newBody)
        Effect.persist(evt).thenRun { _ ⇒
          replyTo ! Done
        }
      case Publish(replyTo) ⇒
        Effect.persist(Published(state.postId)).thenRun { _ ⇒
          println(s"Blog post ${state.postId} was published")
          replyTo ! Done
        }
      case GetPost(replyTo) ⇒
        replyTo ! state.content.get
        Effect.none
      case _: AddPost ⇒
        Effect.unhandled
      case PassivatePost ⇒
        Effect.stop
    }
}
Full source at GitHub
Java
private CommandHandlerBuilder<BlogCommand, BlogEvent, DraftState, BlogState> draftCommandHandler() {
  return commandHandlerBuilder(DraftState.class)
      .matchCommand(ChangeBody.class, (state, cmd) -> {
        BodyChanged event = new BodyChanged(state.postId(), cmd.newBody);
        return Effect().persist(event).andThen(() -> cmd.replyTo.tell(Done.getInstance()));
      })
      .matchCommand(Publish.class, (state, cmd) -> Effect()
          .persist(new Published(state.postId())).andThen(() -> {
            System.out.println("Blog post published: " + state.postId());
            cmd.replyTo.tell(Done.getInstance());
          }))
      .matchCommand(GetPost.class, (state, cmd) -> {
        cmd.replyTo.tell(state.postContent);
        return Effect().none();
      });
}

private CommandHandlerBuilder<BlogCommand, BlogEvent, PublishedState, BlogState> publishedCommandHandler() {
  return commandHandlerBuilder(PublishedState.class)
      .matchCommand(ChangeBody.class, (state, cmd) -> {
        BodyChanged event = new BodyChanged(state.postId(), cmd.newBody);
        return Effect().persist(event).andThen(() -> cmd.replyTo.tell(Done.getInstance()));
      })
      .matchCommand(GetPost.class, (state, cmd) -> {
        cmd.replyTo.tell(state.postContent);
        return Effect().none();
      });
}

private CommandHandlerBuilder<BlogCommand, BlogEvent, BlogState, BlogState> commonCommandHandler() {
  return commandHandlerBuilder(BlogState.class)
      .matchCommand(AddPost.class, (state, cmd) -> Effect().unhandled())
      .matchCommand(PassivatePost.class, (state, cmd) -> Effect().stop());
}
Full source at GitHub

The event handler is always the same independent of state. The main reason for not making the event handler part of the CommandHandler is that contrary to Commands, all events must be handled and that is typically independent of what the current state is. The event handler can still decide what to do based on the state, if that is needed.

Scala
private val eventHandler: (BlogState, BlogEvent) ⇒ BlogState = { (state, event) ⇒
  event match {
    case PostAdded(postId, content) ⇒
      state.withContent(content)

    case BodyChanged(_, newBody) ⇒
      state.content match {
        case Some(c) ⇒ state.copy(content = Some(c.copy(body = newBody)))
        case None    ⇒ state
      }

    case Published(_) ⇒
      state.copy(published = true)
  }
}
Full source at GitHub
Java
@Override
public EventHandler<BlogState, BlogEvent> eventHandler() {
  return eventHandlerBuilder()
      .matchEvent(PostAdded.class, (state, event) ->
          new DraftState(event.content, false))
      .matchEvent(BodyChanged.class, DraftState.class, (state, chg) ->
          state.withContent(new PostContent(state.postId(), state.postContent.title, chg.newBody)))
      .matchEvent(BodyChanged.class, PublishedState.class, (state, chg) ->
          state.withContent(new PostContent(state.postId(), state.postContent.title, chg.newBody)))
      .matchEvent(Published.class, DraftState.class, (state, event) ->
          new PublishedState(state.postContent))
      .build();
}
Full source at GitHub

And finally the behavior is created from the PersistentBehaviors.receive:

Scala
def behavior(entityId: String): Behavior[BlogCommand] =
  PersistentBehaviors.receive[BlogCommand, BlogEvent, BlogState](
    persistenceId = "Blog-" + entityId,
    emptyState = BlogState.empty,
    commandHandler,
    eventHandler)
Full source at GitHub
Java
public static class BlogBehavior extends PersistentBehavior<BlogCommand, BlogEvent, BlogState> {
  public static Behavior<BlogCommand> behavior(String entityId) {
    return Behaviors.setup(ctx ->
        new BlogBehavior("Blog-" + entityId, ctx)
    );
  }

  @Override
  public BlogState emptyState() {
    return new BlankState();
  }

  // commandHandler, eventHandler as in above snippets
}
Full source at GitHub

Effects and Side Effects

Each command has a single Effect which can be:

  • Persist events
  • None: Accept the comment but no effects
  • Unhandled: Don’t handle this message

Note that there is only one of these. It is not possible to both persist and say none/unhandled. These are created using a factory that is returned via the Effect() method the Effect factory and once created additional SideEffects can be added.

Most of them time this will be done with the thenRun method on the Effect above. It is also possible factor out common SideEffects. For example:

Scala
// Example factoring out a chained effect rather than using `andThen`
val commonChainedEffects = SideEffect[Mood](_ ⇒ println("Command processed"))
// Then in a command handler:
Effect.persist(Remembered("Yep")) // persist event
  .andThen(commonChainedEffects) // add on common chained effect
Full source at GitHub
Java
// Factored out Chained effect
static final SideEffect<ExampleState>  commonChainedEffect = SideEffect.create(s -> System.out.println("Command handled!"));

 return commandHandlerBuilder(ExampleState.class)
   .matchCommand(Cmd.class, (state, cmd) -> Effect().persist(new Evt(cmd.data))
     .andThen(() -> cmd.sender.tell(new Ack()))
     .andThen(commonChainedEffect)
   )
   .build();
Full source at GitHub

Side effects ordering and guarantees

Any SideEffects are executed on an at-once basis and will not be executed if the persist fails. The SideEffects are executed sequentially, it is not possible to execute SideEffects in parallel.

Serialization

The same serialization mechanism as for untyped actors is also used in Akka Typed, also for persistent actors. When picking serialization solution for the events you should also consider that it must be possible read old events when the application has evolved. Strategies for that can be found in the schema evolution.

Recovery

It is strongly discouraged to perform side effects in applyEvent, so side effects should be performed once recovery has completed in the onRecoveryCompleted callback. by overriding onRecoveryCompleted

Scala
val recoveryBehavior: Behavior[Command] =
  PersistentBehaviors.receive[Command, Event, State](
    persistenceId = "abc",
    emptyState = State(),
    commandHandler =
      (ctx, state, cmd) ⇒
        throw new RuntimeException("TODO: process the command & return an Effect"),
    eventHandler =
      (state, evt) ⇒
        throw new RuntimeException("TODO: process the event return the next state")
  ).onRecoveryCompleted { (ctx, state) ⇒
      throw new RuntimeException("TODO: add some end-of-recovery side-effect here")
    }
Full source at GitHub
Java
@Override
public void onRecoveryCompleted(State state) {
  throw new RuntimeException("TODO: add some end-of-recovery side-effect here");
}
Full source at GitHub

The onRecoveryCompleted takes an ActorContext and the current State, and doesn’t return anything.

Tagging

Persistence typed allows you to use event tags without using EventAdapter:

Scala
val taggingBehavior: Behavior[Command] =
  PersistentBehaviors.receive[Command, Event, State](
    persistenceId = "abc",
    emptyState = State(),
    commandHandler =
      (ctx, state, cmd) ⇒
        throw new RuntimeException("TODO: process the command & return an Effect"),
    eventHandler =
      (state, evt) ⇒
        throw new RuntimeException("TODO: process the event return the next state")
  ).withTagger(_ ⇒ Set("tag1", "tag2"))
Full source at GitHub
Java
@Override
public Set<String> tagsFor(Event event) {
  throw new RuntimeException("TODO: inspect the event and return any tags it should have");
}
Full source at GitHub

Event adapters

Event adapters can be programmatically added to your PersistentBehaviors that can convert from your Event type to another type that is then passed to the journal.

Defining an event adapter is done by extending an EventAdapter:

Scala
case class Wrapper[T](t: T)
class WrapperEventAdapter[T] extends EventAdapter[T, Wrapper[T]] {
  override def toJournal(e: T): Wrapper[T] = Wrapper(e)
  override def fromJournal(p: Wrapper[T]): T = p.t
}
Full source at GitHub
Java
public static class Wrapper<T> {
  private final T t;
  public Wrapper(T t) {
    this.t = t;
  }
  public T getT() {
    return t;
  }
}

public static class EventAdapterExample extends EventAdapter<SimpleEvent, Wrapper<SimpleEvent>> {
  @Override
  public Wrapper<SimpleEvent> toJournal(SimpleEvent simpleEvent) {
    return new Wrapper<>(simpleEvent);
  }
  @Override
  public SimpleEvent fromJournal(Wrapper<SimpleEvent> simpleEventWrapper) {
    return simpleEventWrapper.getT();
  }
}
Full source at GitHub

Then install it on a persistent behavior:

Scala
persistentBehavior.eventAdapter(new WrapperEventAdapter[Event]))
Full source at GitHub
Java
@Override
public EventAdapter<SimpleEvent, Wrapper<SimpleEvent>> eventAdapter() {
  return new EventAdapterExample();
}
Full source at GitHub

Wrapping Persistent Behaviors

When creating a PersistentBehavior, it is possible to wrap PersistentBehavior in other behaviors such as Behaviors.setup in order to access the ActorContext object. For instance to access the actor logging upon taking snapshots for debug purpose.

Scala
val samplePersistentBehavior = PersistentBehaviors.receive[Command, Event, State](
  persistenceId = "abc",
  emptyState = State(),
  commandHandler =
    (ctx, state, cmd) ⇒
      throw new RuntimeException("TODO: process the command & return an Effect"),
  eventHandler =
    (state, evt) ⇒
      throw new RuntimeException("TODO: process the event return the next state")
).onRecoveryCompleted { (ctx, state) ⇒
    throw new RuntimeException("TODO: add some end-of-recovery side-effect here")
  }

val debugAlwaysSnapshot: Behavior[Command] = Behaviors.setup {
  context ⇒
    samplePersistentBehavior.snapshotWhen((state, _, _) ⇒ {
      context.log.info(
        "Snapshot actor {} => state: {}",
        context.self.path.name, state)
      true
    })
}
Full source at GitHub
Java
static Behavior<Command> debugAlwaysSnapshot = Behaviors.setup((context) -> {
          return new MyPersistentBehavior("pid") {
            @Override
            public boolean shouldSnapshot(State state, Event event, long sequenceNr) {
              context.getLog().info("Snapshot actor {} => state: {}",
                      context.getSelf().path().name(), state);
              return true;
            }
          };
        }
);
Full source at GitHub

Journal failures

By default a PersistentBehavior will stop if an exception is thrown from the journal. It is possible to override this with any BackoffSupervisorStrategy. It is not possible to use the normal supervision wrapping for this as it isn’t valid to resume a behavior on a journal failure as it is not known if the event was persisted.

Scala
val supervisedBehavior = samplePersistentBehavior.onPersistFailure(
  SupervisorStrategy.restartWithBackoff(
    minBackoff = 10.seconds,
    maxBackoff = 60.seconds,
    randomFactor = 0.1
  ))
Full source at GitHub
Java
public static class MyPersistentBehavior extends PersistentBehavior<Command, Event, State> {
  public MyPersistentBehavior(String persistenceId) {
    super(persistenceId, SupervisorStrategy.restartWithBackoff(Duration.ofSeconds(10), Duration.ofSeconds(30), 0.2));
  }
Full source at GitHub

Journal rejections

Journals can reject events. The difference from a failure is that the journal must decide to reject an event before trying to persist it e.g. because of a serialization exception. If an event is rejected it definitely won’t be in the journal. This is signalled to a PersistentBehavior via a EventRejectedException and can be handled with a supervisor.

Found an error in this documentation? The source code for this page can be found here. Please feel free to edit and contribute a pull request.