Configuration

Connection configuration

Shared configuration for the connection pool is located under akka.persistence.r2dbc.connection-factory. You have to set at least:

Postgres

sourceakka.persistence.r2dbc {
  dialect = "postgres"
  connection-factory {
    driver = "postgres"
    host = "localhost"
    host = ${?DB_HOST}
    database = "postgres"
    database = ${?DB_NAME}
    user = "postgres"
    user = ${?DB_USER}
    password = "postgres"
    password = ${?DB_PASSWORD}

    # ssl {
    #   enabled = on
    #   mode = "VERIFY_CA"
    #   root-cert = "/path/db_root.crt"
    # }
  }
}

Yugabyte

sourceakka.persistence.r2dbc {
  dialect = "yugabyte"
  connection-factory {
    driver = "postgres"
    host = "localhost"
    host = ${?DB_HOST}
    port = 5433
    database = "yugabyte"
    database = ${?DB_NAME}
    user = "yugabyte"
    host = ${?DB_USER}
    password = "yugabyte"
    password = ${?DB_PASSWORD}

    # ssl {
    #   enabled = on
    #   mode = "VERIFY_CA"
    #   root-cert = "/path/db_root.crt"
    # }
  }
}

The following can be overridden in your application.conf for the connection settings:

sourceakka.persistence.r2dbc {

  # postgres or yugabyte
  dialect = "postgres"

  # set this to your database schema if applicable, empty by default
  schema = ""

  connection-factory {
    driver = "postgres"

    # the connection can be configured with a url, eg: "r2dbc:postgresql://<host>:5432/<database>"
    url = ""

    # The connection options to be used. Ignored if 'url' is non-empty
    host = "localhost"
    port = 5432
    database = "postgres"
    user = "postgres"
    password = "postgres"

    ssl {
      enabled = off
      # See PostgresqlConnectionFactoryProvider.SSL_MODE
      # Possible values:
      #  allow - encryption if the server insists on it
      #  prefer - encryption if the server supports it
      #  require - encryption enabled and required, but trust network to connect to the right server
      #  verify-ca - encryption enabled and required, and verify server certificate
      #  verify-full - encryption enabled and required, and verify server certificate and hostname
      #  tunnel - use a SSL tunnel instead of following Postgres SSL handshake protocol
      mode = ""

      # Server root certificate. Can point to either a resource within the classpath or a file.
      root-cert = ""

      # Client certificate. Can point to either a resource within the classpath or a file.
      cert = ""

      # Key for client certificate. Can point to either a resource within the classpath or a file.
      key = ""

      # Password for client key.
      password = ""
    }

    # Initial pool size.
    initial-size = 5
    # Maximum pool size.
    max-size = 20
    # Maximum time to create a new connection.
    connect-timeout = 3 seconds
    # Maximum time to acquire connection from pool.
    acquire-timeout = 5 seconds
    # Number of retries if the connection acquisition attempt fails.
    # In the case the database server was restarted all connections in the pool will
    # be invalid. To recover from that without failed acquire you can use the same number
    # of retries as max-size of the pool
    acquire-retry = 1

    # Maximum idle time of the connection in the pool.
    # Background eviction interval of idle connections is derived from this property
    # and max-life-time.
    max-idle-time = 30 minutes

    # Maximum lifetime of the connection in the pool.
    # Background eviction interval of connections is derived from this property
    # and max-idle-time.
    max-life-time = 60 minutes

    # Configures the statement cache size.
    # 0 means no cache, negative values will select an unbounded cache
    # a positive value will configure a bounded cache with the passed size.
    statement-cache-size = 5000

    # Validate the connection when acquired with this SQL.
    # Enabling this has some performance overhead.
    # A fast query for Postgres is "SELECT 1"
    validation-query = ""
  }

  # If database timestamp is guaranteed to not move backwards for two subsequent
  # updates of the same persistenceId there might be a performance gain to
  # set this to `on`. Note that many databases use the system clock and that can
  # move backwards when the system clock is adjusted.
  db-timestamp-monotonic-increasing = off

  # Enable this for testing or workaround of https://github.com/yugabyte/yugabyte-db/issues/10995
  # FIXME: This property will be removed when the Yugabyte issue has been resolved.
  use-app-timestamp = off

  # Logs database calls that take longer than this duration at INFO level.
  # Set to "off" to disable this logging.
  # Set to 0 to log all calls.
  log-db-calls-exceeding = 300 ms

}

Journal configuration

Journal configuration properties are by default defined under akka.persistence.r2dbc.journal.

See Journal plugin configuration.

Snapshot configuration

Snapshot store configuration properties are by default defined under akka.persistence.r2dbc.snapshot.

See Snapshot store plugin configuration.

Durable state configuration

Durable state store configuration properties are by default defined under akka.persistence.r2dbc.state.

See Durable state plugin configuration.

Query configuration

Query configuration properties are by default defined under akka.persistence.r2dbc.query.

See Query plugin configuration.

Multiple plugins

To enable the plugins to be used by default, add the following lines to your Akka application.conf:

sourceakka.persistence.journal.plugin = "akka.persistence.r2dbc.journal"
akka.persistence.snapshot-store.plugin = "akka.persistence.r2dbc.snapshot"
akka.persistence.state.plugin = "akka.persistence.r2dbc.state"

Note that all plugins have a shared root config section akka.persistence.r2dbc, which also contains the Connection configuration for the connection pool that is shared for the plugins.

You can use additional plugins with different configuration. For example if more than one database is used. Then you would define the configuration such as:

sourcesecond-r2dbc = ${akka.persistence.r2dbc}
second-r2dbc {
  connection-factory {
    # specific connection properties here
  }
  journal {
    # specific journal properties here
  }
  snapshot {
    # specific snapshot properties here
  }
  state {
    # specific durable state properties here
  }
  query {
    # specific query properties here
  }
}

To use the additional plugin you would defineoverride the plugin id.

Scala

sourceEventSourcedBehavior(persistenceId, emptyState = State(), commandHandler, eventHandler)
  .withJournalPluginId("second-r2dbc.journal")
  .withSnapshotPluginId("second-r2dbc.snapshot")

Java

sourcepublic class MyEntity extends EventSourcedBehavior<MyEntity.Command, MyEntity.Event, MyEntity.State> {
  @Override
  public String journalPluginId() {
    return "second-r2dbc.journal";
  }

  @Override
  public String snapshotPluginId() {
    return "second-r2dbc.snapshot";
  }
}

It is similar for DurableStateBehavior, define withDurableStateStorePluginId("second-r2dbc.state") override durableStateStorePluginId with "second-r2dbc.state".

For queries and Projection SourceProvider you would use "second-r2dbc.query" instead of the default R2dbcReadJournal.Identifier R2dbcReadJournal.Identifier().

For additional details on multiple plugin configuration for projections see the Akka R2DBC projection docs