CODEWITHSUNDEEP
scalacqrssnapshotevent-sourcingakka-persistence
KaC
KaC

Reputation: 613

Is it possible to rebuild snapshots when state class changes in akka-persistence?

We use EventSourcedBehavior from akka-persistence 2.6.15 for an CQRS/EventSourcing application, with akka-persistence-jdbc 4.0.0 for storing events and snapshots in PostgreSQL DB.

We have state classes that we serialize with snapshots. But sometimes those state classes change, which makes reading snapshot fail obviously. We manage it by deleting those changed snapshots:

      delete from snapshot sn 
      where sn.persistence_id::uuid in (select id from some_entity_table);

But for entities with a lot of events, when sending a new command, it takes a lot of time to get to the latest snapshot, resulting in timeout.

Would it be possible to force a rebuild of snapshots at the startup of the application?

Upvotes: 0

Views: 257

Answers (1)

Levi Ramsey
Levi Ramsey

Reputation: 20561

Arguably the "true" solution to this is to use a SerializerWithStringManifest which can deserialize the previous snapshot formats into the current format. If using reflection-driven serialization, this may be more difficult. It also won't migrate the existing snapshots.

One trick you can do is add an explicit ForceSnapshot command into your persistent actor's protocol. This is pretty easy in the Classic API, where you have more control over snapshotting, so I won't go over that. In the EventSourcedBehavior Typed API, however, this requires a bit more subtlety with adding a SnapshotForced event and changing the snapshotWhen function to return true if the event is a SnapshotForced.

Whether Classic or Typed, you can then have your application force every persistenceId to write a new snapshot by taking advantage of the currentPersistenceIds Persistence Query:

val readJournal =
  PersistenceQuery(system).readJournalFor[JdbcReadJournal](JdbcReadJournal.Identifier)

val everythingSnapshotted =
  readJournal.currentPersistenceIds()
    .mapAsync(parallelism) { id =>
      // use the ask pattern to send a `ForceSnapshot` command and wait for reply
      ???
    }
    .runWith(Sink.ignore)

Upvotes: 0

Related Questions