Persistence (Java with Lambda Support)

Identifiers

A persistent actor must have an identifier that doesn't change across different actor incarnations. The identifier must be defined with the persistenceId method.

@Override
public String persistenceId() {
  return "my-stable-persistence-id";
}

Recovery

By default, a persistent actor is automatically recovered on start and on restart by replaying journaled messages. New messages sent to a persistent actor during recovery do not interfere with replayed messages. New messages will only be received by a persistent actor after recovery completes.

@Override
public Recovery recovery() {
  return Recovery.create(457L);
}

@Override
public Recovery recovery() {
  return Recovery.none();
}

public boolean recoveryRunning();
public boolean recoveryFinished();

class MyPersistentActor5 extends AbstractPersistentActor {

  @Override public String persistenceId() {
    return "my-stable-persistence-id";
  }

  @Override public PartialFunction<Object, BoxedUnit> receiveRecover() {
    return ReceiveBuilder.
      match(RecoveryCompleted.class, r -> {
        // perform init after recovery, before any other messages
        // ...
      }).
      match(String.class, this::handleEvent).build();
  }

  @Override public PartialFunction<Object, BoxedUnit> receiveCommand() {
    return ReceiveBuilder.
      match(String.class, s -> s.equals("cmd"),
        s -> persist("evt", this::handleEvent)).build();
  }

  private void handleEvent(String event) {
    // update state
    // ...
  }

}

Internal stash

The persistent actor has a private stash for internally caching incoming messages during recovery or the persist\persistAll method persisting events. You can still use/inherit from the Stash interface. The internal stash cooperates with the normal stash by hooking into unstashAll method and making sure messages are unstashed properly to the internal stash to maintain ordering guarantees.

You should be careful to not send more messages to a persistent actor than it can keep up with, otherwise the number of stashed messages will grow without bounds. It can be wise to protect against OutOfMemoryError by defining a maximum stash capacity in the mailbox configuration:

akka.actor.default-mailbox.stash-capacity=10000

Note that the stash capacity is per actor. If you have many persistent actors, e.g. when using cluster sharding, you may need to define a small stash capacity to ensure that the total number of stashed messages in the system don't consume too much memory. Additionally, The persistent actor defines three strategies to handle failure when the internal stash capacity is exceeded. The default overflow strategy is the ThrowOverflowExceptionStrategy, which discards the current received message and throws a StashOverflowException, causing actor restart if default supervision strategy is used. you can override the internalStashOverflowStrategy method to return DiscardToDeadLetterStrategy or ReplyToStrategy for any "individual" persistent actor, or define the "default" for all persistent actors by providing FQCN, which must be a subclass of StashOverflowStrategyConfigurator, in the persistence configuration:

akka.persistence.internal-stash-overflow-strategy=
  "akka.persistence.ThrowExceptionConfigurator"

The DiscardToDeadLetterStrategy strategy also has a pre-packaged companion configurator akka.persistence.DiscardConfigurator.

You can also query default strategy via the Akka persistence extension singleton:

Persistence.get(context().system()).defaultInternalStashOverflowStrategy();

Relaxed local consistency requirements and high throughput use-cases

If faced with relaxed local consistency requirements and high throughput demands sometimes PersistentActor and its persist may not be enough in terms of consuming incoming Commands at a high rate, because it has to wait until all Events related to a given Command are processed in order to start processing the next Command. While this abstraction is very useful for most cases, sometimes you may be faced with relaxed requirements about consistency – for example you may want to process commands as fast as you can, assuming that the Event will eventually be persisted and handled properly in the background, retroactively reacting to persistence failures if needed.

The persistAsync method provides a tool for implementing high-throughput persistent actors. It will not stash incoming Commands while the Journal is still working on persisting and/or user code is executing event callbacks.

In the below example, the event callbacks may be called "at any time", even after the next Command has been processed. The ordering between events is still guaranteed ("evt-b-1" will be sent after "evt-a-2", which will be sent after "evt-a-1" etc.).

class MyPersistentActor extends AbstractPersistentActor {

  @Override public String persistenceId() {
    return "my-stable-persistence-id";
  }

  private void handleCommand(String c) {
    sender().tell(c, self());

    persistAsync(String.format("evt-%s-1", c), e -> {
      sender().tell(e, self());
    });
    persistAsync(String.format("evt-%s-2", c), e -> {
      sender().tell(e, self());
    });
  }

  @Override public PartialFunction<Object, BoxedUnit> receiveRecover() {
    return ReceiveBuilder.
      match(String.class, this::handleCommand).build();
  }

  @Override public PartialFunction<Object, BoxedUnit> receiveCommand() {
    return ReceiveBuilder.
      match(String.class, this::handleCommand).build();
  }
}

Deferring actions until preceding persist handlers have executed

Sometimes when working with persistAsync you may find that it would be nice to define some actions in terms of ''happens-after the previous persistAsync handlers have been invoked''. PersistentActor provides an utility method called deferAsync, which works similarly to persistAsync yet does not persist the passed in event. It is recommended to use it for read operations, and actions which do not have corresponding events in your domain model.

Using this method is very similar to the persist family of methods, yet it does not persist the passed in event. It will be kept in memory and used when invoking the handler.

class MyPersistentActor extends AbstractPersistentActor {

  @Override public String persistenceId() {
    return "my-stable-persistence-id";
  }

  private void handleCommand(String c) {
    persistAsync(String.format("evt-%s-1", c), e -> {
      sender().tell(e, self());
    });
    persistAsync(String.format("evt-%s-2", c), e -> {
      sender().tell(e, self());
    });

    deferAsync(String.format("evt-%s-3", c), e -> {
      sender().tell(e, self());
    });
  }

  @Override public PartialFunction<Object, BoxedUnit> receiveRecover() {
    return ReceiveBuilder.
      match(String.class, this::handleCommand).build();
  }

  @Override public PartialFunction<Object, BoxedUnit> receiveCommand() {
    return ReceiveBuilder.
      match(String.class, this::handleCommand).build();
  }
}

Notice that the sender() is safe to access in the handler callback, and will be pointing to the original sender of the command for which this deferAsync handler was called.

final ActorRef persistentActor = system.actorOf(Props.create(MyPersistentActor.class));
persistentActor.tell("a", sender);
persistentActor.tell("b", sender);

// order of received messages:
// a
// b
// evt-a-1
// evt-a-2
// evt-a-3
// evt-b-1
// evt-b-2
// evt-b-3

Nested persist calls

It is possible to call persist and persistAsync inside their respective callback blocks and they will properly retain both the thread safety (including the right value of sender()) as well as stashing guarantees.

In general it is encouraged to create command handlers which do not need to resort to nested event persisting, however there are situations where it may be useful. It is important to understand the ordering of callback execution in those situations, as well as their implication on the stashing behaviour (that persist() enforces). In the following example two persist calls are issued, and each of them issues another persist inside its callback:

@Override public PartialFunction<Object, BoxedUnit> receiveRecover() {
  final Procedure<String> replyToSender = event -> sender().tell(event, self());

  return ReceiveBuilder
    .match(String.class, msg -> {
      persist(String.format("%s-outer-1", msg), event -> {
        sender().tell(event, self());
        persist(String.format("%s-inner-1", event), replyToSender);
      });

      persist(String.format("%s-outer-2", msg), event -> {
        sender().tell(event, self());
        persist(String.format("%s-inner-2", event), replyToSender);
      });
    })
    .build();
}

When sending two commands to this PersistentActor, the persist handlers will be executed in the following order:

persistentActor.tell("a", ActorRef.noSender());
persistentActor.tell("b", ActorRef.noSender());

// order of received messages:
// a
// a-outer-1
// a-outer-2
// a-inner-1
// a-inner-2
// and only then process "b"
// b
// b-outer-1
// b-outer-2
// b-inner-1
// b-inner-2

First the "outer layer" of persist calls is issued and their callbacks are applied. After these have successfully completed, the inner callbacks will be invoked (once the events they are persisting have been confirmed to be persisted by the journal). Only after all these handlers have been successfully invoked will the next command be delivered to the persistent Actor. In other words, the stashing of incoming commands that is guaranteed by initially calling persist() on the outer layer is extended until all nested persist callbacks have been handled.

It is also possible to nest persistAsync calls, using the same pattern:

@Override public PartialFunction<Object, BoxedUnit> receiveRecover() {
    final Procedure<String> replyToSender = event -> sender().tell(event, self());

  return ReceiveBuilder
    .match(String.class, msg -> {
      persistAsync(String.format("%s-outer-1", msg ), event -> {
        sender().tell(event, self());
        persistAsync(String.format("%s-inner-1", event), replyToSender);
      });

      persistAsync(String.format("%s-outer-2", msg ), event -> {
        sender().tell(event, self());
        persistAsync(String.format("%s-inner-1", event), replyToSender);
      });
    })
    .build();
}

In this case no stashing is happening, yet the events are still persisted and callbacks executed in the expected order:

persistentActor.tell("a", self());
persistentActor.tell("b", self());

// order of received messages:
// a
// b
// a-outer-1
// a-outer-2
// b-outer-1
// b-outer-2
// a-inner-1
// a-inner-2
// b-inner-1
// b-inner-2

// which can be seen as the following causal relationship:
// a -> a-outer-1 -> a-outer-2 -> a-inner-1 -> a-inner-2
// b -> b-outer-1 -> b-outer-2 -> b-inner-1 -> b-inner-2

While it is possible to nest mixed persist and persistAsync with keeping their respective semantics it is not a recommended practice, as it may lead to overly complex nesting.

Failures

If persistence of an event fails, onPersistFailure will be invoked (logging the error by default), and the actor will unconditionally be stopped.

The reason that it cannot resume when persist fails is that it is unknown if the even was actually persisted or not, and therefore it is in an inconsistent state. Restarting on persistent failures will most likely fail anyway, since the journal is probably unavailable. It is better to stop the actor and after a back-off timeout start it again. The akka.pattern.BackoffSupervisor actor is provided to support such restarts.

@Override
public void preStart() throws Exception {
  final Props childProps = Props.create(MyPersistentActor1.class);
  final Props props = BackoffSupervisor.props(
    childProps,
    "myActor",
    Duration.create(3, TimeUnit.SECONDS),
    Duration.create(30, TimeUnit.SECONDS),
    0.2);
  context().actorOf(props, "mySupervisor");
  super.preStart();
}

If persistence of an event is rejected before it is stored, e.g. due to serialization error, onPersistRejected will be invoked (logging a warning by default), and the actor continues with next message.

If there is a problem with recovering the state of the actor from the journal when the actor is started, onRecoveryFailure is called (logging the error by default), and the actor will be stopped.

Atomic writes

Each event is of course stored atomically, but it is also possible to store several events atomically by using the persistAll or persistAllAsync method. That means that all events passed to that method are stored or none of them are stored if there is an error.

The recovery of a persistent actor will therefore never be done partially with only a subset of events persisted by persistAll.

Some journals may not support atomic writes of several events and they will then reject the persistAll command, i.e. onPersistRejected is called with an exception (typically UnsupportedOperationException).

Batch writes

In order to optimize throughput when using persistAsync, a persistent actor internally batches events to be stored under high load before writing them to the journal (as a single batch). The batch size is dynamically determined by how many events are emitted during the time of a journal round-trip: after sending a batch to the journal no further batch can be sent before confirmation has been received that the previous batch has been written. Batch writes are never timer-based which keeps latencies at a minimum.

Message deletion

It is possible to delete all messages (journaled by a single persistent actor) up to a specified sequence number; Persistent actors may call the deleteMessages method to this end.

Deleting messages in event sourcing based applications is typically either not used at all or used in conjunction with snapshotting, i.e. after a snapshot has been successfully stored, a deleteMessages(toSequenceNr) up until the sequence number of the data held by that snapshot can be issued to safely delete the previous events while still having access to the accumulated state during replays - by loading the snapshot.

The result of the deleteMessages request is signaled to the persistent actor with a DeleteMessagesSuccess message if the delete was successful or a DeleteMessagesFailure message if it failed.

Persistence status handling

Persisting, deleting and replaying messages can either succeed or fail.

The most important operations (persist and recovery) have failure handlers modelled as explicit callbacks which the user can override in the PersistentActor. The default implementations of these handlers emit a log message (error for persist/recovery failures, and warning for others), logging the failure cause and information about which message caused the failure.

For critical failures, such as recovery or persisting events failing, the persistent actor will be stopped after the failure handler is invoked. This is because if the underlying journal implementation is signalling persistence failures it is most likely either failing completely or overloaded and restarting right-away and trying to persist the event again will most likely not help the journal recover – as it would likely cause a Thundering herd problem, as many persistent actors would restart and try to persist their events again. Instead, using a BackoffSupervisor (as described in Failures) which implements an exponential-backoff strategy which allows for more breathing room for the journal to recover between restarts of the persistent actor.

Safely shutting down persistent actors

Special care should be given when when shutting down persistent actors from the outside. With normal Actors it is often acceptable to use the special PoisonPill message to signal to an Actor that it should stop itself once it receives this message – in fact this message is handled automatically by Akka, leaving the target actor no way to refuse stopping itself when given a poison pill.

This can be dangerous when used with PersistentActor due to the fact that incoming commands are stashed while the persistent actor is awaiting confirmation from the Journal that events have been written when persist() was used. Since the incoming commands will be drained from the Actor's mailbox and put into its internal stash while awaiting the confirmation (thus, before calling the persist handlers) the Actor may receive and (auto)handle the PoisonPill before it processes the other messages which have been put into its stash, causing a pre-mature shutdown of the Actor.

The example below highlights how messages arrive in the Actor's mailbox and how they interact with its internal stashing mechanism when persist() is used. Notice the early stop behaviour that occurs when PoisonPill is used:

final class Shutdown {
}

class MyPersistentActor extends AbstractPersistentActor {
  @Override
  public String persistenceId() {
    return "some-persistence-id";
  }

  @Override
  public PartialFunction<Object, BoxedUnit> receiveCommand() {
    return ReceiveBuilder
      .match(Shutdown.class, shutdown -> {
        context().stop(self());
      })
      .match(String.class, msg -> {
        System.out.println(msg);
        persist("handle-" + msg, e -> System.out.println(e));
      })
      .build();
  }

  @Override
  public PartialFunction<Object, BoxedUnit> receiveRecover() {
    return ReceiveBuilder.matchAny(any -> {}).build();
  }

}

// UN-SAFE, due to PersistentActor's command stashing:
persistentActor.tell("a", ActorRef.noSender());
persistentActor.tell("b", ActorRef.noSender());
persistentActor.tell(PoisonPill.getInstance(), ActorRef.noSender());
// order of received messages:
// a
//   # b arrives at mailbox, stashing;        internal-stash = [b]
//   # PoisonPill arrives at mailbox, stashing; internal-stash = [b, Shutdown]
// PoisonPill is an AutoReceivedMessage, is handled automatically
// !! stop !!
// Actor is stopped without handling `b` nor the `a` handler!

// SAFE:
persistentActor.tell("a", ActorRef.noSender());
persistentActor.tell("b", ActorRef.noSender());
persistentActor.tell(new Shutdown(), ActorRef.noSender());
// order of received messages:
// a
//   # b arrives at mailbox, stashing;        internal-stash = [b]
//   # Shutdown arrives at mailbox, stashing; internal-stash = [b, Shutdown]
// handle-a
//   # unstashing;                            internal-stash = [Shutdown]
// b
// handle-b
//   # unstashing;                            internal-stash = []
// Shutdown
// -- stop --

Updates

The default update interval of all persistent views of an actor system is configurable:

akka.persistence.view.auto-update-interval = 5s

AbstractPersistentView implementation classes may also override the autoUpdateInterval method to return a custom update interval for a specific view class or view instance. Applications may also trigger additional updates at any time by sending a view an Update message.

final ActorRef view = system.actorOf(Props.create(MyView.class));
view.tell(Update.create(true), null);

If the await parameter is set to true, messages that follow the Update request are processed when the incremental message replay, triggered by that update request, completed. If set to false (default), messages following the update request may interleave with the replayed message stream. Automated updates always run with await = false.

Automated updates of all persistent views of an actor system can be turned off by configuration:

akka.persistence.view.auto-update = off

Implementation classes may override the configured default value by overriding the autoUpdate method. To limit the number of replayed messages per update request, applications can configure a custom akka.persistence.view.auto-update-replay-max value or override the autoUpdateReplayMax method. The number of replayed messages for manual updates can be limited with the replayMax parameter of the Update message.

Recovery

Initial recovery of persistent views works the very same way as for persistent actors (i.e. by sending a Recover message to self). The maximum number of replayed messages during initial recovery is determined by autoUpdateReplayMax. Further possibilities to customize initial recovery are explained in section Recovery.

Identifiers

A persistent view must have an identifier that doesn't change across different actor incarnations. The identifier must be defined with the viewId method.

The viewId must differ from the referenced persistenceId, unless Snapshots of a view and its persistent actor should be shared (which is what applications usually do not want).

Snapshot deletion

A persistent actor can delete individual snapshots by calling the deleteSnapshot method with the sequence number of when the snapshot was taken.

To bulk-delete a range of snapshots matching SnapshotSelectionCriteria, persistent actors should use the deleteSnapshots method.

Snapshot status handling

Saving or deleting snapshots can either succeed or fail – this information is reported back to the persistent actor via status messages as illustrated in the following table.

Relationship between deliver and confirmDelivery

To send messages to the destination path, use the deliver method after you have persisted the intent to send the message.

The destination actor must send back a confirmation message. When the sending actor receives this confirmation message you should persist the fact that the message was delivered successfully and then call the confirmDelivery method.

If the persistent actor is not currently recovering, the deliver method will send the message to the destination actor. When recovering, messages will be buffered until they have been confirmed using confirmDelivery. Once recovery has completed, if there are outstanding messages that have not been confirmed (during the message replay), the persistent actor will resend these before sending any other messages.

Deliver requires a deliveryIdToMessage function to pass the provided deliveryId into the message so that the correlation between deliver and confirmDelivery is possible. The deliveryId must do the round trip. Upon receipt of the message, the destination actor will send the same``deliveryId`` wrapped in a confirmation message back to the sender. The sender will then use it to call confirmDelivery method to complete the delivery routine.

class Msg implements Serializable {
  private static final long serialVersionUID = 1L;
  public final long deliveryId;
  public final String s;

  public Msg(long deliveryId, String s) {
    this.deliveryId = deliveryId;
    this.s = s;
  }
}

class Confirm implements Serializable {
  private static final long serialVersionUID = 1L;
  public final long deliveryId;

  public Confirm(long deliveryId) {
    this.deliveryId = deliveryId;
  }
}


class MsgSent implements Serializable {
  private static final long serialVersionUID = 1L;
  public final String s;

  public MsgSent(String s) {
    this.s = s;
  }
}
class MsgConfirmed implements Serializable {
  private static final long serialVersionUID = 1L;
  public final long deliveryId;

  public MsgConfirmed(long deliveryId) {
    this.deliveryId = deliveryId;
  }
}

class MyPersistentActor extends AbstractPersistentActorWithAtLeastOnceDelivery {
  private final ActorSelection destination;

  public MyPersistentActor(ActorSelection destination) {
      this.destination = destination;
  }

  @Override public String persistenceId() {
    return "persistence-id";
  }

  @Override
  public PartialFunction<Object, BoxedUnit> receiveCommand() {
    return ReceiveBuilder.
      match(String.class, s -> {
        persist(new MsgSent(s), evt -> updateState(evt));
      }).
      match(Confirm.class, confirm -> {
        persist(new MsgConfirmed(confirm.deliveryId), evt -> updateState(evt));
      }).
      build();
  }

  @Override
  public PartialFunction<Object, BoxedUnit> receiveRecover() {
    return ReceiveBuilder.
        match(Object.class, evt -> updateState(evt)).build();
  }

  void updateState(Object event) {
    if (event instanceof MsgSent) {
      final MsgSent evt = (MsgSent) event;
      deliver(destination, deliveryId -> new Msg(deliveryId, evt.s));
    } else if (event instanceof MsgConfirmed) {
      final MsgConfirmed evt = (MsgConfirmed) event;
      confirmDelivery(evt.deliveryId);
    }
  }
}

class MyDestination extends AbstractActor {
  public MyDestination() {
    receive(ReceiveBuilder.
      match(Msg.class, msg -> {
        // ...
        sender().tell(new Confirm(msg.deliveryId), self());
      }).build()
    );
  }
}

The deliveryId generated by the persistence module is a strictly monotonically increasing sequence number without gaps. The same sequence is used for all destinations of the actor, i.e. when sending to multiple destinations the destinations will see gaps in the sequence. It is not possible to use custom deliveryId. However, you can send a custom correlation identifier in the message to the destination. You must then retain a mapping between the internal deliveryId (passed into the deliveryIdToMessage function) and your custom correlation id (passed into the message). You can do this by storing such mapping in a Map(correlationId -> deliveryId) from which you can retrieve the deliveryId to be passed into the confirmDelivery method once the receiver of your message has replied with your custom correlation id.

The AbstractPersistentActorWithAtLeastOnceDelivery class has a state consisting of unconfirmed messages and a sequence number. It does not store this state itself. You must persist events corresponding to the deliver and confirmDelivery invocations from your PersistentActor so that the state can be restored by calling the same methods during the recovery phase of the PersistentActor. Sometimes these events can be derived from other business level events, and sometimes you must create separate events. During recovery, calls to deliver will not send out messages, those will be sent later if no matching confirmDelivery will have been performed.

Support for snapshots is provided by getDeliverySnapshot and setDeliverySnapshot. The AtLeastOnceDeliverySnapshot contains the full delivery state, including unconfirmed messages. If you need a custom snapshot for other parts of the actor state you must also include the AtLeastOnceDeliverySnapshot. It is serialized using protobuf with the ordinary Akka serialization mechanism. It is easiest to include the bytes of the AtLeastOnceDeliverySnapshot as a blob in your custom snapshot.

The interval between redelivery attempts is defined by the redeliverInterval method. The default value can be configured with the akka.persistence.at-least-once-delivery.redeliver-interval configuration key. The method can be overridden by implementation classes to return non-default values.

After a number of delivery attempts a AtLeastOnceDelivery.UnconfirmedWarning message will be sent to self. The re-sending will still continue, but you can choose to call confirmDelivery to cancel the re-sending. The number of delivery attempts before emitting the warning is defined by the warnAfterNumberOfUnconfirmedAttempts method. The default value can be configured with the akka.persistence.at-least-once-delivery.warn-after-number-of-unconfirmed-attempts configuration key. The method can be overridden by implementation classes to return non-default values.

The AbstractPersistentActorWithAtLeastOnceDelivery class holds messages in memory until their successful delivery has been confirmed. The maximum number of unconfirmed messages that the actor is allowed to hold in memory is defined by the maxUnconfirmedMessages method. If this limit is exceed the deliver method will not accept more messages and it will throw AtLeastOnceDelivery.MaxUnconfirmedMessagesExceededException. The default value can be configured with the akka.persistence.at-least-once-delivery.max-unconfirmed-messages configuration key. The method can be overridden by implementation classes to return non-default values.

A Simple Example

To demonstrate the features of the AbstractPersistentFSM, consider an actor which represents a Web store customer. The contract of our "WebStoreCustomerFSMActor" is that it accepts the following commands:

AddItem sent when the customer adds an item to a shopping cart Buy - when the customer finishes the purchase Leave - when the customer leaves the store without purchasing anything GetCurrentCart allows to query the current state of customer's shopping cart

The customer can be in one of the following states:

LookingAround customer is browsing the site, but hasn't added anything to the shopping cart Shopping customer has recently added items to the shopping cart Inactive customer has items in the shopping cart, but hasn't added anything recently Paid customer has purchased the items

Customer's actions are "recorded" as a sequence of "domain events" which are persisted. Those events are replayed on actor's start in order to restore the latest customer's state:

Customer state data represents the items in customer's shopping cart:

Here is how everything is wired together:

andThen can be used to define actions which will be executed following event's persistence - convenient for "side effects" like sending a message or logging. Notice that actions defined in andThen block are not executed on recovery:

A snapshot of state data can be persisted by calling the saveStateSnapshot() method:

On recovery state data is initialized according to the latest available snapshot, then the remaining domain events are replayed, triggering the applyEvent method.

Eager initialization of persistence plugin

By default, persistence plugins are started on-demand, as they are used. In some case, however, it might be beneficial to start a certain plugin eagerly. In order to do that, you should first add the akka.persistence.Persistence under the akka.extensions key. Then, specify the IDs of plugins you wish to start automatically under akka.persistence.journal.auto-start-journals and akka.persistence.snapshot-store.auto-start-snapshot-stores.

Journal plugin API

A journal plugin extends AsyncWriteJournal.

AsyncWriteJournal is an actor and the methods to be implemented are:

If the storage backend API only supports synchronous, blocking writes, the methods should be implemented as:

@Override
public Future<Iterable<Optional<Exception>>> doAsyncWriteMessages(
    Iterable<AtomicWrite> messages) {
  try {
    Iterable<Optional<Exception>> result = new ArrayList<Optional<Exception>>();
    // blocking call here...
    // result.add(..)
    return Futures.successful(result);
  } catch (Exception e) {
    return Futures.failed(e);
  }
}

A journal plugin must also implement the methods defined in AsyncRecovery for replays and sequence number recovery:

A journal plugin can be activated with the following minimal configuration:

# Path to the journal plugin to be used
akka.persistence.journal.plugin = "my-journal"

# My custom journal plugin
my-journal {
  # Class name of the plugin.
  class = "docs.persistence.MyJournal"
  # Dispatcher for the plugin actor.
  plugin-dispatcher = "akka.actor.default-dispatcher"
}

The specified plugin class must have a no-arg constructor. The plugin-dispatcher is the dispatcher used for the plugin actor. If not specified, it defaults to akka.persistence.dispatchers.default-plugin-dispatcher.

The journal plugin instance is an actor so the methods corresponding to requests from persistent actors are executed sequentially. It may delegate to asynchronous libraries, spawn futures, or delegate to other actors to achive parallelism.

The journal plugin class must have a constructor without parameters or a constructor with one com.typesafe.config.Config parameter. The plugin section of the actor system's config will be passed in the config constructor parameter.

Don't run journal tasks/futures on the system default dispatcher, since that might starve other tasks.

Snapshot store plugin API

A snapshot store plugin must extend the SnapshotStore actor and implement the following methods:

A snapshot store plugin can be activated with the following minimal configuration:

# Path to the snapshot store plugin to be used
akka.persistence.snapshot-store.plugin = "my-snapshot-store"

# My custom snapshot store plugin
my-snapshot-store {
  # Class name of the plugin.
  class = "docs.persistence.MySnapshotStore"
  # Dispatcher for the plugin actor.
  plugin-dispatcher = "akka.persistence.dispatchers.default-plugin-dispatcher"
}

The specified plugin class must have a no-arg constructor. The plugin-dispatcher is the dispatcher used for the plugin actor. If not specified, it defaults to akka.persistence.dispatchers.default-plugin-dispatcher.

The snapshot store instance is an actor so the methods corresponding to requests from persistent actors are executed sequentially. It may delegate to asynchronous libraries, spawn futures, or delegate to other actors to achive parallelism.

The snapshot store plugin class must have a constructor without parameters or constructor with one com.typesafe.config.Config parameter. The plugin section of the actor system's config will be passed in the config constructor parameter.

Don't run snapshot store tasks/futures on the system default dispatcher, since that might starve other tasks.

Local LevelDB journal

The LevelDB journal plugin config entry is akka.persistence.journal.leveldb. It writes messages to a local LevelDB instance. Enable this plugin by defining config property:

# Path to the journal plugin to be used
akka.persistence.journal.plugin = "akka.persistence.journal.leveldb"

LevelDB based plugins will also require the following additional dependency declaration:

<dependency>
  <groupId>org.iq80.leveldb</groupId>
  <artifactId>leveldb</artifactId>
  <version>0.7</version>
</dependency>
<dependency>
  <groupId>org.fusesource.leveldbjni</groupId>
  <artifactId>leveldbjni-all</artifactId>
  <version>1.8</version>
</dependency>

The default location of LevelDB files is a directory named journal in the current working directory. This location can be changed by configuration where the specified path can be relative or absolute:

akka.persistence.journal.leveldb.dir = "target/journal"

With this plugin, each actor system runs its own private LevelDB instance.

Shared LevelDB journal

A LevelDB instance can also be shared by multiple actor systems (on the same or on different nodes). This, for example, allows persistent actors to failover to a backup node and continue using the shared journal instance from the backup node.

A shared LevelDB instance is started by instantiating the SharedLeveldbStore actor.

final ActorRef store = system.actorOf(Props.create(SharedLeveldbStore.class), "store");

By default, the shared instance writes journaled messages to a local directory named journal in the current working directory. The storage location can be changed by configuration:

akka.persistence.journal.leveldb-shared.store.dir = "target/shared"

Actor systems that use a shared LevelDB store must activate the akka.persistence.journal.leveldb-shared plugin.

akka.persistence.journal.plugin = "akka.persistence.journal.leveldb-shared"

This plugin must be initialized by injecting the (remote) SharedLeveldbStore actor reference. Injection is done by calling the SharedLeveldbJournal.setStore method with the actor reference as argument.

class SharedStorageUsage extends UntypedActor {
  @Override
  public void preStart() throws Exception {
    String path = "akka.tcp://example@127.0.0.1:2552/user/store";
    ActorSelection selection = getContext().actorSelection(path);
    selection.tell(new Identify(1), getSelf());
  }

  @Override
  public void onReceive(Object message) throws Exception {
    if (message instanceof ActorIdentity) {
      ActorIdentity identity = (ActorIdentity) message;
      if (identity.correlationId().equals(1)) {
        ActorRef store = identity.getRef();
        if (store != null) {
          SharedLeveldbJournal.setStore(store, getContext().system());
        }
      }
    }
  }
}

Internal journal commands (sent by persistent actors) are buffered until injection completes. Injection is idempotent i.e. only the first injection is used.

Local snapshot store

Local snapshot store plugin config entry is akka.persistence.snapshot-store.local. It writes snapshot files to the local filesystem. Enable this plugin by defining config property:

# Path to the snapshot store plugin to be used
akka.persistence.snapshot-store.plugin = "akka.persistence.snapshot-store.local"

The default storage location is a directory named snapshots in the current working directory. This can be changed by configuration where the specified path can be relative or absolute:

akka.persistence.snapshot-store.local.dir = "target/snapshots"

Note that it is not mandatory to specify a snapshot store plugin. If you don't use snapshots you don't have to configure it.

Persistence Plugin Proxy

A persistence plugin proxy allows sharing of journals and snapshot stores across multiple actor systems (on the same or on different nodes). This, for example, allows persistent actors to failover to a backup node and continue using the shared journal instance from the backup node. The proxy works by forwarding all the journal/snapshot store messages to a single, shared, persistence plugin instance, and therefor supports any use case supported by the proxied plugin.

The journal and snapshot store proxies are controlled via the akka.persistence.journal.proxy and akka.persistence.snapshot-store.proxy configuration entries, respectively. Set the target-journal-plugin or target-snapshot-store-plugin keys to the underlying plugin you wish to use (for example: akka.persistence.journal.leveldb). The start-target-journal and start-target-snapshot-store keys should be set to on in exactly one actor system - this is the system that will instantiate the shared persistence plugin. Next, the proxy needs to be told how to find the shared plugin. This can be done by setting the target-journal-address and target-snapshot-store-address configuration keys, or programmatically by calling the PersistencePluginProxy.setTargetLocation method.