Cassandra Plugins for Akka Persistence

Replicated Akka Persistence journal and snapshot store backed by Apache Cassandra.

Dependencies

Latest release

To include the latest release of the Cassandra plugins into your sbt project, add the following lines to your build.sbt file:

libraryDependencies += "com.typesafe.akka" %% "akka-persistence-cassandra" % "0.22"

This version of akka-persistence-cassandra depends on Akka 2.4.14. It has been published for Scala 2.11 and 2.12.

It is compatible with Cassandra 3.0.0 or higher, and it is also compatible with Cassandra 2.1.6 or higher (versions < 2.1.6 have a static column bug) if you configure cassandra-journal.cassandra-2x-compat=on in your application.conf. With Cassandra 2.x compatibility some features will not be enabled, e.g. eventsByTag.

It implements the following Persistence Queries:

• allPersistenceIds, currentPersistenceIds
• eventsByPersistenceId, currentEventsByPersistenceId
• eventsByTag, currentEventsByTag (only for Cassandra 3.x)

Migrations

Migrations from 0.11 to 0.12

Dispatcher configuration was changed, see reference.conf:

Migrations from 0.9 to 0.10

The event data, snapshot data and meta data are stored in a separate columns instead of being wrapped in blob. Run the following statements in cqlsh:

drop materialized view akka.eventsbytag;
alter table akka.messages add writer_uuid text;
alter table akka.messages add ser_id int;
alter table akka.messages add ser_manifest text;
alter table akka.messages add event_manifest text;
alter table akka.messages add event blob;
alter table akka_snapshot.snapshots add ser_id int;
alter table akka_snapshot.snapshots add ser_manifest text;
alter table akka_snapshot.snapshots add snapshot_data blob;

Migrations from 0.6 to 0.7

Schema changes mean that you can't upgrade from version 0.6 for Cassandra 2.x of the plugin to the 0.7 version and use existing data without schema migration. You should be able to export the data and load it to the new table definition.

Migrating from 0.3.x (Akka 2.3.x)

Schema and property changes mean that you can't currently upgrade from 0.3 to 0.4 SNAPSHOT and use existing data without schema migration. You should be able to export the data and load it to the new table definition.

Journal plugin

Features

• All operations required by the Akka Persistence journal plugin API are fully supported.
• The plugin uses Cassandra in a pure log-oriented way i.e. data are only ever inserted but never updated (deletions are made on user request only).
• Writes of messages are batched to optimize throughput for persistAsync. See batch writes for details how to configure batch sizes. The plugin was tested to work properly under high load.
• Messages written by a single persistent actor are partitioned across the cluster to achieve scalability with data volume by adding nodes.
• Persistence Query support by CassandraReadJournal

Configuration

To activate the journal plugin, add the following line to your Akka application.conf:

akka.persistence.journal.plugin = "cassandra-journal"

This will run the journal with its default settings. The default settings can be changed with the configuration properties defined in reference.conf:

Caveats

• Detailed tests under failure conditions are still missing.
• Range deletion performance (i.e. deleteMessages up to a specified sequence number) depends on the extend of previous deletions
  • linearly increases with the number of tombstones generated by previous permanent deletions and drops to a minimum after compaction

These issues are likely to be resolved in future versions of the plugin.

Persistence Query usage example

Obtain a stream with all events tagged with "someTag" with Persistence Query:

val queries = PersistenceQuery(system).readJournalFor[CassandraReadJournal](CassandraReadJournal.Identifier)
queries.eventsByTag("someTag", readJournal.firstOffset)

Snapshot store plugin

Features

• Implements the Akka Persistence snapshot store plugin API.

Configuration

To activate the snapshot-store plugin, add the following line to your Akka application.conf:

akka.persistence.snapshot-store.plugin = "cassandra-snapshot-store"

This will run the snapshot store with its default settings. The default settings can be changed with the configuration properties defined in reference.conf: