Offset in a relational DB with Slick
The SlickProjection
SlickProjection
has support for storing the offset in a relational database using Slick (JDBC). This is only an option for Scala and for Java the offset can be stored in relational DB with JDBC. The JDBC module can also be used with Scala.
The Slick module in Akka Projections is community-driven and not included in Lightbend support. Prefer using the JDBC module to implement your projection handler. Slick support in Akka Projections is meant for users migrating from Lagom's Slick ReadSideProcessor
.
The source of the envelopes can be events from Akka Persistence or any other SourceProvider
with supported offset types.
The envelope handler returns a DBIO
that will be run by the projection. This means that the target database operations can be run in the same transaction as the storage of the offset, which means that exactly-once processing semantics is supported. It also offers at-least-once semantics.
Dependencies
The Akka dependencies are available from Akka’s library repository. To access them there, you need to configure the URL for this repository.
- sbt
resolvers += "Akka library repository".at("https://repo.akka.io/maven")
- Maven
<project> ... <repositories> <repository> <id>akka-repository</id> <name>Akka library repository</name> <url>https://repo.akka.io/maven</url> </repository> </repositories> </project>
- Gradle
repositories { mavenCentral() maven { url "https://repo.akka.io/maven" } }
To use the Slick module of Akka Projections add the following dependency in your project:
- sbt
libraryDependencies += "com.lightbend.akka" %% "akka-projection-slick" % "1.5.7"
- Maven
<properties> <scala.binary.version>2.13</scala.binary.version> </properties> <dependencies> <dependency> <groupId>com.lightbend.akka</groupId> <artifactId>akka-projection-slick_${scala.binary.version}</artifactId> <version>1.5.7</version> </dependency> </dependencies>
- Gradle
def versions = [ ScalaBinary: "2.13" ] dependencies { implementation "com.lightbend.akka:akka-projection-slick_${versions.ScalaBinary}:1.5.7" }
Akka Projections require Akka 2.9.6 or later, see Akka version.
Project Info: Akka Projections Slick | |
---|---|
Artifact | com.lightbend.akka
akka-projection-slick
1.5.7
|
JDK versions | Eclipse Temurin JDK 11 Eclipse Temurin JDK 17 Eclipse Temurin JDK 21 |
Scala versions | 2.13.15, 3.3.4 |
JPMS module name | akka.projection.slick |
License | |
Readiness level |
Since 0.0, 2020-04-01
|
Home page | https://akka.io |
API documentation | |
Forums | |
Release notes | GitHub releases |
Issues | GitHub issues |
Sources | https://github.com/akka/akka-projection |
Transitive dependencies
The table below shows akka-projection-slick
’s direct dependencies and the second tab shows all libraries it depends on transitively.
- Direct dependencies
Organization Artifact Version com.lightbend.akka akka-projection-core_2.13 1.5.7 com.lightbend.akka akka-projection-jdbc_2.13 1.5.7 com.typesafe.akka akka-persistence-query_2.13 2.9.6 com.typesafe.slick slick_2.13 3.5.2 org.scala-lang scala-library 2.13.15 org.slf4j slf4j-api 1.7.36 - Dependency tree
com.lightbend.akka akka-projection-core_2.13 1.5.7 com.typesafe.akka akka-actor-typed_2.13 2.9.6 BUSL-1.1 com.typesafe.akka akka-actor_2.13 2.9.6 BUSL-1.1 com.typesafe config 1.4.3 Apache-2.0 org.scala-lang.modules scala-java8-compat_2.13 1.0.2 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 com.typesafe.akka akka-slf4j_2.13 2.9.6 BUSL-1.1 com.typesafe.akka akka-actor_2.13 2.9.6 BUSL-1.1 com.typesafe config 1.4.3 Apache-2.0 org.scala-lang.modules scala-java8-compat_2.13 1.0.2 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.slf4j slf4j-api 1.7.36 org.scala-lang scala-library 2.13.15 Apache-2.0 org.slf4j slf4j-api 1.7.36 com.typesafe.akka akka-persistence-query_2.13 2.9.6 BUSL-1.1 com.typesafe.akka akka-persistence_2.13 2.9.6 BUSL-1.1 com.typesafe.akka akka-actor_2.13 2.9.6 BUSL-1.1 com.typesafe config 1.4.3 Apache-2.0 org.scala-lang.modules scala-java8-compat_2.13 1.0.2 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 com.typesafe.akka akka-stream_2.13 2.9.6 BUSL-1.1 com.typesafe.akka akka-actor_2.13 2.9.6 BUSL-1.1 com.typesafe config 1.4.3 Apache-2.0 org.scala-lang.modules scala-java8-compat_2.13 1.0.2 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 com.typesafe.akka akka-protobuf-v3_2.13 2.9.6 BUSL-1.1 org.reactivestreams reactive-streams 1.0.4 MIT-0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 com.typesafe.akka akka-protobuf-v3_2.13 2.9.6 BUSL-1.1 com.typesafe.akka akka-stream_2.13 2.9.6 BUSL-1.1 com.typesafe.akka akka-actor_2.13 2.9.6 BUSL-1.1 com.typesafe config 1.4.3 Apache-2.0 org.scala-lang.modules scala-java8-compat_2.13 1.0.2 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 com.typesafe.akka akka-protobuf-v3_2.13 2.9.6 BUSL-1.1 org.reactivestreams reactive-streams 1.0.4 MIT-0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 com.typesafe.akka akka-protobuf-v3_2.13 2.9.6 BUSL-1.1 com.typesafe.akka akka-stream_2.13 2.9.6 BUSL-1.1 com.typesafe.akka akka-actor_2.13 2.9.6 BUSL-1.1 com.typesafe config 1.4.3 Apache-2.0 org.scala-lang.modules scala-java8-compat_2.13 1.0.2 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 com.typesafe.akka akka-protobuf-v3_2.13 2.9.6 BUSL-1.1 org.reactivestreams reactive-streams 1.0.4 MIT-0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 com.lightbend.akka akka-projection-jdbc_2.13 1.5.7 com.lightbend.akka akka-projection-core_2.13 1.5.7 com.typesafe.akka akka-actor-typed_2.13 2.9.6 BUSL-1.1 com.typesafe.akka akka-actor_2.13 2.9.6 BUSL-1.1 com.typesafe config 1.4.3 Apache-2.0 org.scala-lang.modules scala-java8-compat_2.13 1.0.2 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 com.typesafe.akka akka-slf4j_2.13 2.9.6 BUSL-1.1 com.typesafe.akka akka-actor_2.13 2.9.6 BUSL-1.1 com.typesafe config 1.4.3 Apache-2.0 org.scala-lang.modules scala-java8-compat_2.13 1.0.2 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.slf4j slf4j-api 1.7.36 org.scala-lang scala-library 2.13.15 Apache-2.0 org.slf4j slf4j-api 1.7.36 com.typesafe.akka akka-persistence-query_2.13 2.9.6 BUSL-1.1 com.typesafe.akka akka-persistence_2.13 2.9.6 BUSL-1.1 com.typesafe.akka akka-actor_2.13 2.9.6 BUSL-1.1 com.typesafe config 1.4.3 Apache-2.0 org.scala-lang.modules scala-java8-compat_2.13 1.0.2 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 com.typesafe.akka akka-stream_2.13 2.9.6 BUSL-1.1 com.typesafe.akka akka-actor_2.13 2.9.6 BUSL-1.1 com.typesafe config 1.4.3 Apache-2.0 org.scala-lang.modules scala-java8-compat_2.13 1.0.2 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 com.typesafe.akka akka-protobuf-v3_2.13 2.9.6 BUSL-1.1 org.reactivestreams reactive-streams 1.0.4 MIT-0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 com.typesafe.akka akka-protobuf-v3_2.13 2.9.6 BUSL-1.1 com.typesafe.akka akka-stream_2.13 2.9.6 BUSL-1.1 com.typesafe.akka akka-actor_2.13 2.9.6 BUSL-1.1 com.typesafe config 1.4.3 Apache-2.0 org.scala-lang.modules scala-java8-compat_2.13 1.0.2 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 com.typesafe.akka akka-protobuf-v3_2.13 2.9.6 BUSL-1.1 org.reactivestreams reactive-streams 1.0.4 MIT-0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 com.typesafe.akka akka-protobuf-v3_2.13 2.9.6 BUSL-1.1 com.typesafe.akka akka-stream_2.13 2.9.6 BUSL-1.1 com.typesafe.akka akka-actor_2.13 2.9.6 BUSL-1.1 com.typesafe config 1.4.3 Apache-2.0 org.scala-lang.modules scala-java8-compat_2.13 1.0.2 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 com.typesafe.akka akka-protobuf-v3_2.13 2.9.6 BUSL-1.1 org.reactivestreams reactive-streams 1.0.4 MIT-0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 com.typesafe.akka akka-persistence-query_2.13 2.9.6 BUSL-1.1 com.typesafe.akka akka-persistence_2.13 2.9.6 BUSL-1.1 com.typesafe.akka akka-actor_2.13 2.9.6 BUSL-1.1 com.typesafe config 1.4.3 Apache-2.0 org.scala-lang.modules scala-java8-compat_2.13 1.0.2 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 com.typesafe.akka akka-stream_2.13 2.9.6 BUSL-1.1 com.typesafe.akka akka-actor_2.13 2.9.6 BUSL-1.1 com.typesafe config 1.4.3 Apache-2.0 org.scala-lang.modules scala-java8-compat_2.13 1.0.2 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 com.typesafe.akka akka-protobuf-v3_2.13 2.9.6 BUSL-1.1 org.reactivestreams reactive-streams 1.0.4 MIT-0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 com.typesafe.akka akka-protobuf-v3_2.13 2.9.6 BUSL-1.1 com.typesafe.akka akka-stream_2.13 2.9.6 BUSL-1.1 com.typesafe.akka akka-actor_2.13 2.9.6 BUSL-1.1 com.typesafe config 1.4.3 Apache-2.0 org.scala-lang.modules scala-java8-compat_2.13 1.0.2 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 com.typesafe.akka akka-protobuf-v3_2.13 2.9.6 BUSL-1.1 org.reactivestreams reactive-streams 1.0.4 MIT-0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 com.typesafe.akka akka-persistence-query_2.13 2.9.6 BUSL-1.1 com.typesafe.akka akka-persistence_2.13 2.9.6 BUSL-1.1 com.typesafe.akka akka-actor_2.13 2.9.6 BUSL-1.1 com.typesafe config 1.4.3 Apache-2.0 org.scala-lang.modules scala-java8-compat_2.13 1.0.2 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 com.typesafe.akka akka-stream_2.13 2.9.6 BUSL-1.1 com.typesafe.akka akka-actor_2.13 2.9.6 BUSL-1.1 com.typesafe config 1.4.3 Apache-2.0 org.scala-lang.modules scala-java8-compat_2.13 1.0.2 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 com.typesafe.akka akka-protobuf-v3_2.13 2.9.6 BUSL-1.1 org.reactivestreams reactive-streams 1.0.4 MIT-0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 com.typesafe.akka akka-protobuf-v3_2.13 2.9.6 BUSL-1.1 com.typesafe.akka akka-stream_2.13 2.9.6 BUSL-1.1 com.typesafe.akka akka-actor_2.13 2.9.6 BUSL-1.1 com.typesafe config 1.4.3 Apache-2.0 org.scala-lang.modules scala-java8-compat_2.13 1.0.2 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 com.typesafe.akka akka-protobuf-v3_2.13 2.9.6 BUSL-1.1 org.reactivestreams reactive-streams 1.0.4 MIT-0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 com.typesafe.slick slick_2.13 3.5.2 Two-clause BSD-style license com.typesafe config 1.4.3 Apache-2.0 org.reactivestreams reactive-streams 1.0.4 MIT-0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.scala-lang scala-library 2.13.15 Apache-2.0 org.slf4j slf4j-api 1.7.36
exactly-once
The offset is stored in the same transaction as the DBIO
returned from the handler
, which means exactly-once processing semantics if the projection is restarted from previously stored offset.
- Scala
-
source
import akka.projection.ProjectionId import akka.projection.slick.SlickProjection import slick.basic.DatabaseConfig import slick.dbio.DBIO import slick.jdbc.H2Profile implicit val ec: ExecutionContext = system.executionContext val projection = SlickProjection.exactlyOnce( projectionId = ProjectionId("ShoppingCarts", "carts-1"), sourceProvider, dbConfig, handler = () => new ShoppingCartHandler(repository))
The ShoppingCartHandler
is shown below.
at-least-once
The offset is stored after the envelope has been processed and giving at-least-once processing semantics. This means that if the projection is restarted from a previously stored offset some elements may be processed more than once. Therefore, the Handler code must be idempotent.
- Scala
-
source
implicit val ec: ExecutionContext = system.executionContext val projection = SlickProjection .atLeastOnce( projectionId = ProjectionId("ShoppingCarts", "carts-1"), sourceProvider, dbConfig, handler = () => new ShoppingCartHandler(repository)) .withSaveOffset(afterEnvelopes = 100, afterDuration = 500.millis)
The offset is stored after a time window, or limited by a number of envelopes, whatever happens first. This window can be defined with withSaveOffset
of the returned AtLeastOnceProjection
. The default settings for the window is defined in configuration section akka.projection.at-least-once
. There is a performance benefit of not storing the offset too often, but the drawback is that there can be more duplicates when the projection that will be processed again when the projection is restarted.
The ShoppingCartHandler
is shown below.
groupedWithin
The envelopes can be grouped before processing, which can be useful for batch updates.
- Scala
-
source
implicit val ec: ExecutionContext = system.executionContext val projection = SlickProjection .groupedWithin( projectionId = ProjectionId("ShoppingCarts", "carts-1"), sourceProvider, dbConfig, handler = () => new GroupedShoppingCartHandler(repository)) .withGroup(groupAfterEnvelopes = 20, groupAfterDuration = 500.millis)
The envelopes are grouped within a time window, or limited by a number of envelopes, whatever happens first. This window can be defined with withGroup
of the returned GroupedProjection
. The default settings for the window is defined in configuration section akka.projection.grouped
.
When using groupedWithin
the handler is a SlickHandler[immutable.Seq[EventEnvelope[ShoppingCart.Event]]]
. The GroupedShoppingCartHandler
is shown below.
The offset is stored in the same transaction as the DBIO
returned from the handler
, which means exactly-once processing semantics if the projection is restarted from previously stored offset.
Handler
It’s in the SlickHandler
SlickHandler
that you implement the processing of each envelope. It’s essentially a function from Envelope
to DBIO[Done]
. The returned DBIO
is run by the projection.
A handler that is consuming ShoppingCart.Event
from eventsByTag
can look like this:
- Scala
-
source
import scala.concurrent.Future import akka.Done import akka.projection.slick.SlickHandler import org.slf4j.LoggerFactory class ShoppingCartHandler(repository: OrderRepository)(implicit ec: ExecutionContext) extends SlickHandler[EventEnvelope[ShoppingCart.Event]] { private val logger = LoggerFactory.getLogger(getClass) override def process(envelope: EventEnvelope[ShoppingCart.Event]): DBIO[Done] = { envelope.event match { case ShoppingCart.CheckedOut(cartId, time) => logger.info(s"Shopping cart $cartId was checked out at $time") repository.save(Order(cartId, time)) case otherEvent => logger.debug(s"Shopping cart ${otherEvent.cartId} changed by $otherEvent") DBIO.successful(Done) } } }
Such simple handlers can also be defined as plain functions via the helper SlickHandler.apply
factory method.
where the OrderRepository
is:
- Scala
-
source
case class Order(id: String, time: Instant) class OrderRepository(val dbConfig: DatabaseConfig[H2Profile]) { import dbConfig.profile.api._ private class OrdersTable(tag: Tag) extends Table[Order](tag, "ORDERS") { def id = column[String]("CART_ID", O.PrimaryKey) def time = column[Instant]("TIME") def * = (id, time).mapTo[Order] } private val ordersTable = TableQuery[OrdersTable] def save(order: Order)(implicit ec: ExecutionContext) = { ordersTable.insertOrUpdate(order).map(_ => Done) } def createTable(): Future[Unit] = dbConfig.db.run(ordersTable.schema.createIfNotExists) }
with the Slick DatabaseConfig
:
- Scala
-
source
val dbConfig: DatabaseConfig[H2Profile] = DatabaseConfig.forConfig("akka.projection.slick", system.settings.config) val repository = new OrderRepository(dbConfig)
Grouped handler
When using SlickProjection.groupedWithin
the handler is processing a Seq
of envelopes.
- Scala
-
source
import scala.collection.immutable class GroupedShoppingCartHandler(repository: OrderRepository)(implicit ec: ExecutionContext) extends SlickHandler[immutable.Seq[EventEnvelope[ShoppingCart.Event]]] { private val logger = LoggerFactory.getLogger(getClass) override def process(envelopes: immutable.Seq[EventEnvelope[ShoppingCart.Event]]): DBIO[Done] = { val dbios = envelopes.map(_.event).map { case ShoppingCart.CheckedOut(cartId, time) => logger.info(s"Shopping cart $cartId was checked out at $time") repository.save(Order(cartId, time)) case otherEvent => logger.debug(s"Shopping cart ${otherEvent.cartId} changed by $otherEvent") DBIO.successful(Done) } DBIO.sequence(dbios).map(_ => Done) } }
Stateful handler
The SlickHandler
can be stateful, with variables and mutable data structures. It is invoked by the Projection
machinery one envelope at a time and visibility guarantees between the invocations are handled automatically, i.e. no volatile or other concurrency primitives are needed for managing the state as long as it’s not accessed by other threads than the one that called process
.
It is important that the Handler
instance is not shared between several Projection
instances, because then it would be invoked concurrently, which is not how it is intended to be used. Each Projection
instance should use a new Handler
instance.
Async handler
The Handler
Handler
can be used with SlickProjection.atLeastOnceAsync
and SlickProjection.groupedWithinAsync
if the handler is not storing the projection result in the database. The handler could send to a Kafka topic or integrate with something else.
There are several examples of such Handler
in the documentation for Cassandra Projections. Same type of handlers can be used with SlickProjection
instead of CassandraProjection
.
Actor handler
A good alternative for advanced state management is to implement the handler as an actor, which is described in Processing with Actor.
Flow handler
An Akka Streams FlowWithContext
can be used instead of a handler for processing the envelopes, which is described in Processing with Akka Streams.
Handler lifecycle
You can override the start
and stop
methods of the SlickHandler
SlickHandler
to implement initialization before first envelope is processed and resource cleanup when the projection is stopped. Those methods are also called when the Projection
is restarted after failure.
See also error handling.
Schema
The database schema for the offset storage table:
- PostgreSQL
-
source
CREATE TABLE IF NOT EXISTS akka_projection_offset_store ( projection_name VARCHAR(255) NOT NULL, projection_key VARCHAR(255) NOT NULL, current_offset VARCHAR(255) NOT NULL, manifest VARCHAR(4) NOT NULL, mergeable BOOLEAN NOT NULL, last_updated BIGINT NOT NULL, PRIMARY KEY(projection_name, projection_key) ); CREATE INDEX IF NOT EXISTS akka_projection_name_index ON akka_projection_offset_store (projection_name); CREATE TABLE IF NOT EXISTS akka_projection_management ( projection_name VARCHAR(255) NOT NULL, projection_key VARCHAR(255) NOT NULL, paused BOOLEAN NOT NULL, last_updated BIGINT NOT NULL, PRIMARY KEY(projection_name, projection_key) );
- MySQL
-
source
CREATE TABLE IF NOT EXISTS akka_projection_offset_store ( projection_name VARCHAR(255) NOT NULL, projection_key VARCHAR(255) NOT NULL, current_offset VARCHAR(255) NOT NULL, manifest VARCHAR(4) NOT NULL, mergeable BOOLEAN NOT NULL, last_updated BIGINT NOT NULL, PRIMARY KEY(projection_name, projection_key) ); CREATE INDEX akka_projection_name_index ON akka_projection_offset_store (projection_name); CREATE TABLE IF NOT EXISTS akka_projection_management ( projection_name VARCHAR(255) NOT NULL, projection_key VARCHAR(255) NOT NULL, paused BOOLEAN NOT NULL, last_updated BIGINT NOT NULL, PRIMARY KEY(projection_name, projection_key) );
- Microsoft SQL Server
-
source
IF NOT EXISTS (SELECT 1 FROM SYS.OBJECTS WHERE OBJECT_ID = OBJECT_ID(N'akka_projection_offset_store') AND TYPE IN (N'U')) BEGIN CREATE TABLE akka_projection_offset_store ( projection_name VARCHAR(255) NOT NULL, projection_key VARCHAR(255) NOT NULL, current_offset VARCHAR(255) NOT NULL, manifest VARCHAR(4) NOT NULL, mergeable BIT NOT NULL, last_updated BIGINT NOT NULL ) ALTER TABLE akka_projection_offset_store ADD CONSTRAINT pk_projection_id PRIMARY KEY(projection_name, projection_key) CREATE INDEX akka_projection_name_index ON akka_projection_offset_store (projection_name) END IF NOT EXISTS (SELECT 1 FROM SYS.OBJECTS WHERE OBJECT_ID = OBJECT_ID(N'akka_projection_management') AND TYPE IN (N'U')) BEGIN CREATE TABLE akka_projection_management ( projection_name VARCHAR(255) NOT NULL, projection_key VARCHAR(255) NOT NULL, paused BIT NOT NULL, last_updated BIGINT NOT NULL ) ALTER TABLE akka_projection_management ADD CONSTRAINT pk_projection_management_id PRIMARY KEY(projection_name, projection_key) END
- Oracle
-
source
BEGIN execute immediate 'create table "AKKA_PROJECTION_OFFSET_STORE" ("PROJECTION_NAME" VARCHAR2(255) NOT NULL,"PROJECTION_KEY" VARCHAR2(255) NOT NULL,"CURRENT_OFFSET" VARCHAR2(255) NOT NULL,"MANIFEST" VARCHAR2(4) NOT NULL,"MERGEABLE" CHAR(1) NOT NULL check ("MERGEABLE" in (0, 1)),"LAST_UPDATED" NUMBER(19) NOT NULL) '; execute immediate 'alter table "AKKA_PROJECTION_OFFSET_STORE" add constraint "PK_PROJECTION_ID" primary key("PROJECTION_NAME","PROJECTION_KEY") '; execute immediate 'create index "AKKA_PROJECTION_NAME_INDEX" on "AKKA_PROJECTION_OFFSET_STORE" ("PROJECTION_NAME") '; EXCEPTION WHEN OTHERS THEN IF SQLCODE = -955 THEN NULL; -- suppresses ORA-00955 exception ELSE RAISE; END IF; END; BEGIN execute immediate 'create table "AKKA_PROJECTION_MANAGEMENT" ("PROJECTION_NAME" VARCHAR2(255) NOT NULL,"PROJECTION_KEY" VARCHAR2(255) NOT NULL,"PAUSED" CHAR(1) NOT NULL check ("PAUSED" in (0, 1)),"LAST_UPDATED" NUMBER(19) NOT NULL) '; execute immediate 'alter table "AKKA_PROJECTION_MANAGEMENT" add constraint "PK_PROJECTION_MANAGEMENT_ID" primary key("PROJECTION_NAME","PROJECTION_KEY") '; EXCEPTION WHEN OTHERS THEN IF SQLCODE = -955 THEN NULL; -- suppresses ORA-00955 exception ELSE RAISE; END IF; END;
- H2
-
source
CREATE TABLE IF NOT EXISTS "akka_projection_offset_store" ( "projection_name" VARCHAR(255) NOT NULL, "projection_key" VARCHAR(255) NOT NULL, "current_offset" VARCHAR(255) NOT NULL, "manifest" VARCHAR(4) NOT NULL, "mergeable" BOOLEAN NOT NULL, "last_updated" BIGINT NOT NULL, PRIMARY KEY("projection_name", "projection_key") ); CREATE INDEX IF NOT EXISTS "akka_projection_name_index" ON "akka_projection_offset_store" ("projection_name"); CREATE TABLE IF NOT EXISTS "akka_projection_management" ( "projection_name" VARCHAR(255) NOT NULL, "projection_key" VARCHAR(255) NOT NULL, "paused" BOOLEAN NOT NULL, "last_updated" BIGINT NOT NULL, PRIMARY KEY("projection_name", "projection_key") );
The schema can be created and dropped using the methods SlickProjection.createTablesIfNotExists
and SlickProjection.dropTablesIfExists
. This is particularly useful when writting tests. For production enviornments, we recommend creating the schema before deploying the application.
As of version 1.1.0, the schema for PostgreSQL and H2 databases has changed. It now defaults to lowercase table and column names. If you have a schema in production, we recommend applying an ALTER table script to change it accordingly.
Alternatively, you can fallback to the uppercase format. You will also need to set akka.projection.slick.offset-store.table
as an uppercase value, as this setting is now defaulting to lowercase.
akka.projection.slick.offset-store {
table = "AKKA_PROJECTION_OFFSET_STORE"
use-lowercase-schema = false
}
Offset types
The supported offset types of the SlickProjection
are:
Offset
Offset
types from events from Akka PersistenceMergeableOffset
MergeableOffset
that is used for messages from KafkaString
Int
Long
- Any other type that has a configured Akka Serializer is stored with base64 encoding of the serialized bytes.
Configuration
Make your edits/overrides in your application.conf.
The reference configuration file with the default values:
sourceakka.projection.slick {
# The Slick profile to use
# set to one of:
# slick.jdbc.H2Profile$
# slick.jdbc.PostgresProfile$
# slick.jdbc.MySQLProfile$
# slick.jdbc.SQLServerProfile$
# slick.jdbc.OracleProfile$
#profile = <fill this with your profile of choice>
# add here your Slick db settings
db {
# url = "jdbc:h2:mem:test1;OPTIMIZE_REUSE_RESULTS=FALSE"
# driver = org.h2.Driver
# connectionPool = disabled
# keepAliveConnection = true
}
offset-store {
# set this to your database schema if applicable, empty by default
schema = ""
# the database table name for the offset store
table = "akka_projection_offset_store"
# the database table name for the projection manangement data
management-table = "akka_projection_management"
# Use lowercase table and column names.
# This is mostly useful for H2 and Postgres databases. MySQL and SQL Server are case insensitive.
# Oracle schema is case sensitive and is defined with uppercase, this property is therefore ignore when using Oracle
use-lowercase-schema = true
}
}