Cassandra snapshot store

build.sbt

@@ -175,7 +175,16 @@ lazy val `tests` = (project in file("tests")
     Slf4j.`log4j-over-slf4j` % Test,
     Logback.core % Test,
     Logback.classic % Test,
-    scalatest % Test)))
+    scalatest % Test,
+    // these libraries are only need for
+    // Akka Persistence Cassandra Persistence performance tests
+    Akka.`persistence-cassandra` % Test,
+    Akka.`akka-cluster` % Test,
+    Akka.`akka-cluster-tools` % Test,
+    Akka.`akka-coordination` % Test,
+    Akka.`akka-persistence-query` % Test,
+    Akka.`akka-pki` % Test,
+    Akka.`akka-remote` % Test)))
 
 lazy val replicator = (Project("replicator", file("replicator"))
   settings (name := "kafka-journal-replicator")

persistence/src/main/resources/reference.conf

@@ -55,6 +55,39 @@ evolutiongaming.kafka-journal.persistence {
     json-codec = default
   }
 
+  snapshot {
+
+    class = "akka.persistence.kafka.journal.CassandraSnapshotStore"
+
+    plugin-dispatcher = "evolutiongaming.kafka-journal.persistence.dispatcher"
+
+    persistence-id-to-key {
+      impl = "constant-topic"  # valid values: constant-topic, split
+      constant-topic {
+        topic = "journal"
+      }
+    }
+
+    cassandra {
+      client {
+        name = "journal"
+        query {
+          fetch-size = 1000
+          consistency = "LOCAL_QUORUM"
+          default-idempotence = true
+        }
+      }
+    }
+
+    circuit-breaker {
+      max-failures = 100
+      call-timeout = 130s // should be higher than producer.delivery-timeout
+      reset-timeout = 1m
+    }
+
+    json-codec = default
+  }
+
   dispatcher {
     type = "Dispatcher"
     executor = "fork-join-executor"
@@ -64,4 +97,4 @@ evolutiongaming.kafka-journal.persistence {
       parallelism-max = 32
     }
   }
-}
+}

persistence/src/main/scala/akka/persistence/kafka/journal/CassandraSnapshotStore.scala

@@ -0,0 +1,201 @@
+package akka.persistence.kafka.journal
+
+import akka.actor.ActorSystem
+import akka.persistence.snapshot.SnapshotStore
+import akka.persistence.{SelectedSnapshot, SnapshotMetadata, SnapshotSelectionCriteria}
+import cats.effect.IO
+import cats.effect.kernel.Resource
+import cats.effect.syntax.all._
+import cats.effect.unsafe.{IORuntime, IORuntimeConfig}
+import cats.syntax.all._
+import com.evolutiongaming.catshelper.CatsHelper._
+import com.evolutiongaming.catshelper.{FromFuture, LogOf, ToFuture}
+import com.evolutiongaming.kafka.journal.snapshot.cassandra.SnapshotCassandraConfig
+import com.evolutiongaming.kafka.journal.util.CatsHelper._
+import com.evolutiongaming.kafka.journal.util.PureConfigHelper._
+import com.evolutiongaming.kafka.journal.{JsonCodec, LogOfFromAkka, Origin, Payload, SnapshotReadWrite}
+import com.evolutiongaming.retry.Retry.implicits._
+import com.evolutiongaming.retry.{OnError, Strategy}
+import com.evolutiongaming.scassandra.CassandraClusterOf
+import com.typesafe.config.Config
+import pureconfig.ConfigSource
+
+import scala.concurrent.duration._
+import scala.concurrent.{Await, ExecutionContextExecutor, Future}
+
+class CassandraSnapshotStore(config: Config) extends SnapshotStore { actor =>
+
+  implicit val system: ActorSystem = context.system
+  implicit val executor: ExecutionContextExecutor = context.dispatcher
+
+  private val (blocking, blockingShutdown) = IORuntime.createDefaultBlockingExecutionContext("kafka-journal-blocking")
+  private val (scheduler, schedulerShutdown) = IORuntime.createDefaultScheduler("kafka-journal-scheduler")
+  implicit val ioRuntime: IORuntime = IORuntime(
+    compute = executor,
+    blocking = blocking,
+    scheduler = scheduler,
+    shutdown = () => {
+      blockingShutdown()
+      schedulerShutdown()
+    },
+    config = IORuntimeConfig()
+  )
+  implicit val toFuture: ToFuture[IO] = ToFuture.ioToFuture
+  implicit val fromFuture: FromFuture[IO] = FromFuture.lift[IO]
+
+  val adapter: Future[(SnapshotStoreAdapter[Future], IO[Unit])] =
+    adapterIO
+      .map { _.mapK(toFuture.toFunctionK, fromFuture.toFunctionK) }
+      .allocated
+      .toFuture
+
+  override def loadAsync(persistenceId: String, criteria: SnapshotSelectionCriteria): Future[Option[SelectedSnapshot]] =
+    adapter.flatMap { case (adapter, _) => adapter.load(persistenceId, criteria) }
+
+  override def saveAsync(metadata: SnapshotMetadata, snapshot: Any): Future[Unit] =
+    adapter.flatMap { case (adapter, _) => adapter.save(metadata, snapshot) }
+
+  override def deleteAsync(metadata: SnapshotMetadata): Future[Unit] =
+    adapter.flatMap { case (adapter, _) => adapter.delete(metadata) }
+
+  override def deleteAsync(persistenceId: String, criteria: SnapshotSelectionCriteria): Future[Unit] =
+    adapter.flatMap { case (adapter, _) => adapter.delete(persistenceId, criteria) }
+
+  override def postStop(): Unit = {
+    val future = adapter.flatMap { case (_, release) => release.toFuture }
+    Await.result(future, 1.minute)
+    super.postStop()
+  }
+
+  def adapterIO: Resource[IO, SnapshotStoreAdapter[IO]] = {
+    for {
+      snapshotSerializer <- serializer
+      config <- cassandraSnapshotStoreConfig.toResource
+      snapshotReadWrite <- snapshotReadWrite(config).toResource
+      adapter <- adapterIO(config, snapshotSerializer, snapshotReadWrite)
+    } yield adapter
+  }
+
+  def adapterIO[A](
+    snapshotSerializer: SnapshotSerializer[IO, A],
+    snapshotReadWrite: SnapshotReadWrite[IO, A]
+  ): Resource[IO, SnapshotStoreAdapter[IO]] = {
+    for {
+      config <- cassandraSnapshotStoreConfig.toResource
+      adapter <- adapterIO(config, snapshotSerializer, snapshotReadWrite)
+    } yield adapter
+  }
+
+  def adapterIO[A](
+    config: CassandraSnapshotStoreConfig,
+    snapshotSerializer: SnapshotSerializer[IO, A],
+    snapshotReadWrite: SnapshotReadWrite[IO, A]
+  ): Resource[IO, SnapshotStoreAdapter[IO]] = {
+    for {
+      logOf <- logOf
+      log <- logOf(classOf[KafkaJournal]).toResource
+      _ <- log.debug(s"config: $config").toResource
+      adapter <- Resource {
+        val adapter = for {
+          toKey <- toKey
+          origin <- origin.toResource
+          cassandraClusterOf <- cassandraClusterOf
+          adapter <- adapterOf(
+            toKey = toKey,
+            origin = origin,
+            snapshotSerializer = snapshotSerializer,
+            snapshotReadWrite = snapshotReadWrite,
+            config = config.cassandra,
+            cassandraClusterOf = cassandraClusterOf
+          )(logOf = logOf)
+        } yield adapter
+        val strategy = Strategy.fibonacci(100.millis).cap(config.startTimeout)
+        val onError: OnError[IO, Throwable] = { (error, status, decision) =>
+          {
+            decision match {
+              case OnError.Decision.Retry(delay) =>
+                log.warn(s"allocate failed, retrying in $delay, error: $error")
+
+              case OnError.Decision.GiveUp =>
+                val retries = status.retries
+                val duration = status.delay
+                log.error(s"allocate failed after $retries retries within $duration: $error", error)
+            }
+          }
+        }
+        adapter.allocated
+          .retry(strategy, onError)
+          .timeout(config.startTimeout)
+          .map { case (adapter, release0) =>
+            val release = release0
+              .timeout(config.startTimeout)
+              .handleErrorWith { error => log.error(s"release failed with $error", error) }
+            (adapter, release)
+          }
+      }
+    } yield adapter
+  }
+
+  def adapterOf[A](
+    toKey: ToKey[IO],
+    origin: Option[Origin],
+    snapshotSerializer: SnapshotSerializer[IO, A],
+    snapshotReadWrite: SnapshotReadWrite[IO, A],
+    config: SnapshotCassandraConfig,
+    cassandraClusterOf: CassandraClusterOf[IO]
+  )(implicit logOf: LogOf[IO]): Resource[IO, SnapshotStoreAdapter[IO]] =
+    SnapshotStoreAdapter.of[IO, A](
+      toKey = toKey,
+      origin = origin,
+      snapshotSerializer = snapshotSerializer,
+      snapshotReadWrite = snapshotReadWrite,
+      config = config,
+      cassandraClusterOf = cassandraClusterOf
+    )
+
+  def toKey: Resource[IO, ToKey[IO]] =
+    ToKey.fromConfig[IO](config).pure[Resource[IO, *]]
+
+  def origin: IO[Option[Origin]] = {
+    val hostName = Origin.hostName[IO]
+    def akkaHost = Origin.akkaHost[IO](system)
+    def akkaName = Origin.akkaName(system)
+    hostName.toOptionT
+      .orElse(akkaHost.toOptionT)
+      .orElse(akkaName.some.toOptionT[IO])
+      .value
+  }
+
+  def cassandraSnapshotStoreConfig: IO[CassandraSnapshotStoreConfig] =
+    ConfigSource
+      .fromConfig(config)
+      .load[CassandraSnapshotStoreConfig]
+      .liftTo[IO]
+
+  def serializer: Resource[IO, SnapshotSerializer[IO, Payload]] =
+    SnapshotSerializer.of[IO](system).toResource
+
+  def snapshotReadWrite(config: CassandraSnapshotStoreConfig): IO[SnapshotReadWrite[IO, Payload]] =
+    for {
+      jsonCodec <- jsonCodec(config)
+    } yield {
+      implicit val jsonCodec1 = jsonCodec
+      SnapshotReadWrite.of[IO, Payload]
+    }
+
+  def jsonCodec(config: CassandraSnapshotStoreConfig): IO[JsonCodec[IO]] = {
+    val codec: JsonCodec[IO] = config.jsonCodec match {
+      case KafkaJournalConfig.JsonCodec.Default  => JsonCodec.default
+      case KafkaJournalConfig.JsonCodec.PlayJson => JsonCodec.playJson
+      case KafkaJournalConfig.JsonCodec.Jsoniter => JsonCodec.jsoniter
+    }
+    codec.pure[IO]
+  }
+
+  def cassandraClusterOf: Resource[IO, CassandraClusterOf[IO]] =
+    CassandraClusterOf.of[IO].toResource
+
+  def logOf: Resource[IO, LogOf[IO]] =
+    LogOfFromAkka[IO](system).pure[Resource[IO, *]]
+
+}

persistence/src/main/scala/akka/persistence/kafka/journal/CassandraSnapshotStoreConfig.scala

@@ -0,0 +1,56 @@
+package akka.persistence.kafka.journal
+
+import com.evolutiongaming.kafka.journal.snapshot.cassandra.SnapshotCassandraConfig
+import pureconfig.generic.semiauto.deriveReader
+import pureconfig.{ConfigCursor, ConfigReader, ConfigSource}
+
+import scala.concurrent.duration._
+
+/** Configuration for [[CassandraSnapshotStore]].
+  *
+  * This case class specifies configuration that could be set using `application.conf` (see `reference.conf` for an
+  * example of such configuration).
+  *
+  * @param cassandra
+  *   Cassandra-specific configuration used by a plugin.
+  * @param startTimeout
+  *   The timeout to create a journal adapter. Starting a journal involves some effectful steps, such as creating
+  *   Cassandra session, so, in case of infrastructure or configuration troubles, it could take a longer time. Creating
+  *   the journal will fail with [[TimeoutException]] if it takes longer than `startTimeout`.
+  * @param stopTimeout
+  *   This is meant to be a counterpart to `startTimeout`, allowing resource release to timeout with an error. This
+  *   parameter is not used, for now, and `startTimeout` is used instead.
+  * @param jsonCodec
+  *   JSON codec to use for (de)serialization of the events from [[scodec.bits.ByteVector]] to
+  *   [[play.api.libs.json.JsValue]] and vice-versa. This parameter is only relevant if default [[CassandraSnapshotStore]] is
+  *   used, i.e. it is not taken into account if Circe JSON or other custom serialization is used.
+  *
+  * @see
+  *   [[KafkaJournal]] for more details.
+  */
+final case class CassandraSnapshotStoreConfig(
+  cassandra: SnapshotCassandraConfig = SnapshotCassandraConfig.default,
+  startTimeout: FiniteDuration = 1.minute,
+  stopTimeout: FiniteDuration = 1.minute,
+  jsonCodec: KafkaJournalConfig.JsonCodec = KafkaJournalConfig.JsonCodec.Default
+)
+
+object CassandraSnapshotStoreConfig {
+
+  val default: CassandraSnapshotStoreConfig = CassandraSnapshotStoreConfig()
+
+  implicit val configReaderKafkaJournalConfig: ConfigReader[CassandraSnapshotStoreConfig] = {
+
+    val configReader = deriveReader[CassandraSnapshotStoreConfig]
+
+    cursor: ConfigCursor => {
+      for {
+        cursor <- cursor.asObjectCursor
+        config = cursor.objValue.toConfig
+        source = ConfigSource.fromConfig(config)
+        config <- source.load(configReader)
+      } yield config
+    }
+  }
+
+}