Migration Tool

There is a migration tool that is useful if you would like to migrate from another Pekko Persistence plugin to the R2DBC plugin. It has been tested with Pekko Persistence JDBC as source plugin, but it should work with any plugin that has support for CurrentPersistenceIdsQuery and CurrentEventsByPersistenceIdQuery.

The migration tool can be run while the source system is still active, and it can be run multiple times with idempotent result. Full rolling update when switching database or Persistence plugin is not supported, but you can migrate most of the data while the system is online and then have a short full shutdown while migrating the remaining data that was written after the previous online migration.

Dependencies

Maven

<properties>
  <scala.binary.version>2.13</scala.binary.version>
</properties>
<dependencies>
  <dependency>
    <groupId>org.apache.pekko</groupId>
    <artifactId>pekko-persistence-r2dbc-migration_${scala.binary.version}</artifactId>
    <version>1.0.0</version>
  </dependency>
</dependencies>

sbt

libraryDependencies += "org.apache.pekko" %% "pekko-persistence-r2dbc-migration" % "1.0.0"

Gradle

def versions = [
  ScalaBinary: "2.13"
]
dependencies {
  implementation "org.apache.pekko:pekko-persistence-r2dbc-migration_${versions.ScalaBinary}:1.0.0"
}

Progress table

To speed up processing of subsequent runs it stores migrated persistence ids and sequence numbers in the table migration_progress. In a subsequent run it will only migrate new events and snapshots compared to what was stored in migration_progress. It will also find and migrate new persistence ids in a subsequent run. You can delete from migration_progress if you want to re-run the full migration.

It’s recommended that you create the migration_progress table before running the migration tool, but if it doesn’t exist the tool will try to create the table.

CREATE TABLE IF NOT EXISTS migration_progress(
  persistence_id VARCHAR(255) NOT NULL,
  event_seq_nr BIGINT,
  snapshot_seq_nr BIGINT,
  PRIMARY KEY(persistence_id)

Configuration

The migration tool can be run as main class org.apache.pekko.persistence.r2dbc.migration.MigrationTool provided by the above pekko-persistence-r2dbc-migration dependency.

You need to provide configuration for the source persistence plugin and the target Rd2BC plugin in your application.conf. An example of such configuration for migration from Pekko Persistence JDBC:

copysource# SPDX-License-Identifier: Apache-2.0

pekko.persistence.r2dbc.migration {
  source {
    query-plugin-id = "jdbc-read-journal"
    snapshot-plugin-id = "jdbc-snapshot-store"
  }
}

pekko.persistence.r2dbc {
  # use different table names or schema
  journal.table = "event_journal2"
  snapshot.table = "snapshot2"
  state.table = "durable_state2"
}

pekko.persistence.r2dbc.connection-factory {
  driver = "postgres"
  host = "localhost"
  port = 5432
  user = "postgres"
  password = "postgres"
  database = "postgres"
}

pekko-persistence-jdbc {
  shared-databases {
    default {
      profile = "slick.jdbc.PostgresProfile$"
      db {
        host = "localhost"
        url = "jdbc:postgresql://localhost:5432/postgres?reWriteBatchedInserts=true"
        user = postgres
        password = postgres
        driver = "org.postgresql.Driver"
        numThreads = 20
        maxConnections = 20
        minConnections = 5
      }
    }
  }
}

jdbc-journal {
  use-shared-db = "default"
}
jdbc-snapshot-store {
  use-shared-db = "default"
}
jdbc-read-journal {
  use-shared-db = "default"
}

# application specific serializers for events and snapshots
# must also be configured and included in classpath

Reference configuration

The following can be overridden in your application.conf for the migration tool specific settings:

copysource# SPDX-License-Identifier: Apache-2.0

pekko.persistence.r2dbc.migration {

  # Pekko Persistence plugin to migrate from.
  # You must also define plugin specific configuration
  # and application specific serializers for events and snapshots.
  source {
    query-plugin-id = "jdbc-read-journal"
    snapshot-plugin-id = "jdbc-snapshot-store"
  }

  # R2DBC Pekko Persistence plugin to migrate to.
  # You must also define pekko-persistence-r2dbc specific configuration.
  target {
    # this must be a configuration path of pekko-persistence-r2dbc
    persistence-plugin-id = "pekko.persistence.r2dbc"

    # Events are stored in batches of this size.
    batch = 10
  }

  # How many persistence ids to migrate concurrently.
  parallelism = 10

}