From 2bbb28d5fc31f62167afd718caf01d90e2382079 Mon Sep 17 00:00:00 2001 From: Domantas Petrauskas Date: Tue, 27 Oct 2026 10:27:47 +0200 Subject: [PATCH 1/3] r2dbc: support H2 database Motivation: Akka Projection 1.5.0 added H2 support to the R2DBC projections (akka/akka-projection#914, akka/akka-projection#938), and apache/pekko-persistence-r2dbc#517 added the h2 dialect to the persistence plugin. Modification: Add H2R2dbcOffsetStore with MERGE upserts and CURRENT_TIMESTAMP. With the h2 dialect the projection creates the offset store tables of its own settings before the first connection, like the persistence plugins, instead of upstream's connection-level additional-init schema. Database calls run on the connection factory's h2.use-dispatcher. The executor reads close-calls-exceeding from the connection factory config, since the snapshot deprecates the constructor without it. Build against the pekko-persistence-r2dbc snapshot that includes apache/pekko-persistence-r2dbc#517, add test-only H2 dependencies, an H2 CI job and docs. Result: R2DBC projections can store offsets in an H2 database, with the tables created automatically. Tests: - r2dbc-int-test/test with h2 dialect: 124 succeeded, 0 failed - r2dbc-int-test/test with PostgreSQL: 124 succeeded, 0 failed - checkCodeStyle; mimaReportBinaryIssues, +headerCheckAll, docs/paradox: pass References: Refs #662 --- .github/workflows/integration-tests-r2dbc.yml | 46 +++++ docs/src/main/paradox/r2dbc.md | 17 ++ .../projection/grpc/TestDbLifecycle.scala | 7 +- project/Dependencies.scala | 7 + project/PekkoPersistenceR2DBCDependency.scala | 2 +- .../r2dbc/H2OffsetStoreTablesSpec.scala | 161 ++++++++++++++++++ .../r2dbc/R2dbcProjectionSpec.scala | 9 +- .../r2dbc/RuntimePluginConfigSpec.scala | 39 ++--- .../pekko/projection/r2dbc/TestConfig.scala | 11 ++ .../projection/r2dbc/TestDbLifecycle.scala | 70 ++++---- r2dbc/src/main/resources/reference.conf | 2 +- .../r2dbc/internal/R2dbcOffsetStore.scala | 3 + .../r2dbc/internal/R2dbcProjectionImpl.scala | 47 ++++- .../internal/h2/H2OffsetStoreTables.scala | 133 +++++++++++++++ .../internal/h2/H2R2dbcOffsetStore.scala | 55 ++++++ .../r2dbc/scaladsl/R2dbcProjection.scala | 44 ++--- 16 files changed, 561 insertions(+), 92 deletions(-) create mode 100644 r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/H2OffsetStoreTablesSpec.scala create mode 100644 r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/h2/H2OffsetStoreTables.scala create mode 100644 r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/h2/H2R2dbcOffsetStore.scala diff --git a/.github/workflows/integration-tests-r2dbc.yml b/.github/workflows/integration-tests-r2dbc.yml index 8b6fd807..17dde9df 100644 --- a/.github/workflows/integration-tests-r2dbc.yml +++ b/.github/workflows/integration-tests-r2dbc.yml @@ -207,3 +207,49 @@ jobs: - name: Print logs on failure if: ${{ failure() }} run: find . -name "*.log" -exec ./scripts/cat-log.sh {} \; + + h2-test: + name: Build and Test Integration for R2DBC with H2 + runs-on: ubuntu-22.04 + if: github.repository == 'apache/pekko-projection' + strategy: + fail-fast: false + matrix: + include: + - { java-version: 17, scala-version: 2.13, sbt-opts: '' } + - { java-version: 17, scala-version: next, sbt-opts: '' } + + steps: + - name: Checkout + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + fetch-depth: 0 + fetch-tags: true + + - name: Checkout GitHub merge + if: github.event.pull_request + run: |- + git fetch origin pull/${{ github.event.pull_request.number }}/merge:scratch + git checkout scratch + + - name: Cache Coursier cache + uses: coursier/cache-action@057d08c2f9a2c4e7a4d33df3aae657b7db88b8cc # v8.2.0 + + - name: Setup JDK ${{ matrix.java-version }} + uses: actions/setup-java@b843ebf9b913146b38da2eaa7aeddf76074ae1ad # v6.1.0 + with: + java-version: ${{ matrix.java-version }} + distribution: temurin + + - name: Install sbt + uses: sbt/setup-sbt@ecad189dd245444b9ae0a0cd5922fd938b88906f # v1.5.12 + + # H2 runs in the test JVM and the projection creates its tables, no database container or schema script + - name: test + run: |- + scala_version="$(bash .github/scripts/resolve-scala-version.sh "${{ matrix.scala-version }}")" + sbt -Dpekko.persistence.r2dbc.dialect=h2 -Dpekko.projection.r2dbc.dialect=h2 "++${scala_version}!" "r2dbc-int-test/test" ${{ matrix.sbt-opts }} + + - name: Print logs on failure + if: ${{ failure() }} + run: find . -name "*.log" -exec ./scripts/cat-log.sh {} \; diff --git a/docs/src/main/paradox/r2dbc.md b/docs/src/main/paradox/r2dbc.md index 5211fa7e..74281775 100644 --- a/docs/src/main/paradox/r2dbc.md +++ b/docs/src/main/paradox/r2dbc.md @@ -54,6 +54,23 @@ YugaByte MySQL : @@snip [MySQL Schema](/r2dbc-int-test/ddl-scripts/create_tables_mysql.sql) +H2 +: The tables are created automatically, see @ref:[H2](#h2). + +### H2 + +With the `h2` dialect the database runs in the same JVM process, see +@extref:[H2 in Apache Pekko Persistence R2DBC](pekko-persistence-r2dbc:getting-started.html#h2) for how to configure the +connection and the H2 dependencies. + +Like the persistence plugins, a projection creates the offset store tables of its own `offset-store` settings when it is +first used, so no schema script is needed. That includes the `schema` and table names of the projection settings and a +separate `use-connection-factory`. The `projection_offset_store` table is not created when `offset-table` is set to +`""`. The tables are not created when the connection is configured with an H2 `url`. + +The database calls of the projection run on the `h2.use-dispatcher` of the connection factory, since the H2 driver +blocks. + ## Configuration By default, `pekko-projection-r2dbc` uses the same connection pool and `dialect` as `pekko-persistence-r2dbc`, see diff --git a/grpc-int-test/src/test/scala/org/apache/pekko/projection/grpc/TestDbLifecycle.scala b/grpc-int-test/src/test/scala/org/apache/pekko/projection/grpc/TestDbLifecycle.scala index 316c334b..6dcfee7d 100644 --- a/grpc-int-test/src/test/scala/org/apache/pekko/projection/grpc/TestDbLifecycle.scala +++ b/grpc-int-test/src/test/scala/org/apache/pekko/projection/grpc/TestDbLifecycle.scala @@ -17,6 +17,7 @@ import org.apache.pekko import pekko.actor.typed.ActorSystem import pekko.persistence.Persistence import pekko.persistence.r2dbc.ConnectionFactoryProvider +import pekko.persistence.r2dbc.ConnectionFactorySettings import pekko.persistence.r2dbc.JournalSettings import pekko.persistence.r2dbc.internal.R2dbcExecutor import pekko.projection.r2dbc.R2dbcProjectionSettings @@ -40,7 +41,11 @@ trait TestDbLifecycle extends BeforeAndAfterAll { this: Suite => new R2dbcExecutor( ConnectionFactoryProvider(typedSystem).connectionFactoryFor(r2dbcProjectionSettings.useConnectionFactory), LoggerFactory.getLogger(getClass), - r2dbcProjectionSettings.logDbCallsExceeding)(typedSystem.executionContext, typedSystem) + r2dbcProjectionSettings.logDbCallsExceeding, + ConnectionFactorySettings.closeCallsExceeding( + typedSystem.settings.config.getConfig(r2dbcProjectionSettings.useConnectionFactory)))( + typedSystem.executionContext, + typedSystem) } lazy val persistenceExt: Persistence = Persistence(typedSystem) diff --git a/project/Dependencies.scala b/project/Dependencies.scala index 4c2c84ab..cd67877b 100644 --- a/project/Dependencies.scala +++ b/project/Dependencies.scala @@ -69,6 +69,9 @@ object Dependencies { val r2dbcPool = "io.r2dbc" % "r2dbc-pool" % "1.0.2.RELEASE" val r2dbcPostgres = "org.postgresql" % "r2dbc-postgresql" % "1.1.3.RELEASE" val r2dbcMysql = "io.asyncer" % "r2dbc-mysql" % "1.4.3" + val r2dbcH2 = "io.r2dbc" % "r2dbc-h2" % "1.1.0.RELEASE" // ApacheV2 + // the H2 version that pekko-persistence-r2dbc is built against + val h2R2dbc = "com.h2database" % "h2" % "2.5.252" // EPL 1.0 or MPL 2.0 } object Test { @@ -114,6 +117,8 @@ object Dependencies { val r2dbcPostgres = Compile.r2dbcPostgres % "test" val r2dbcMysql = Compile.r2dbcMysql % "test" + val r2dbcH2 = Compile.r2dbcH2 % "test" + val h2R2dbc = Compile.h2R2dbc % "test" } object Examples { @@ -276,6 +281,8 @@ object Dependencies { Compile.r2dbcPool, Test.r2dbcPostgres, Test.r2dbcMysql, + Test.r2dbcH2, + Test.h2R2dbc, Test.pekkoSerializationJackson, Test.pekkoDiscovery, Test.pekkoDistributedData, diff --git a/project/PekkoPersistenceR2DBCDependency.scala b/project/PekkoPersistenceR2DBCDependency.scala index 3a3e2835..1a62a6ef 100644 --- a/project/PekkoPersistenceR2DBCDependency.scala +++ b/project/PekkoPersistenceR2DBCDependency.scala @@ -21,5 +21,5 @@ import com.github.pjfanning.pekkobuild.PekkoDependency object PekkoPersistenceR2DBCDependency extends PekkoDependency { override val checkProject: String = "pekko-persistence-r2dbc" override val module: Option[String] = None - override val currentVersion: String = "2.0.0-M1" + override val currentVersion: String = "2.0.0-M1+113-ac55a4d8-SNAPSHOT" } diff --git a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/H2OffsetStoreTablesSpec.scala b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/H2OffsetStoreTablesSpec.scala new file mode 100644 index 00000000..e20bb84a --- /dev/null +++ b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/H2OffsetStoreTablesSpec.scala @@ -0,0 +1,161 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.pekko.projection.r2dbc + +import java.util.UUID + +import org.apache.pekko +import pekko.actor.testkit.typed.scaladsl.LogCapturing +import pekko.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit +import pekko.persistence.r2dbc.ConnectionFactoryProvider +import pekko.persistence.r2dbc.Dialect +import pekko.projection.ProjectionId +import pekko.projection.internal.ManagementState +import pekko.projection.r2dbc.internal.OffsetPidSeqNr +import pekko.projection.r2dbc.internal.R2dbcOffsetStore +import pekko.projection.r2dbc.internal.R2dbcProjectionImpl +import pekko.projection.r2dbc.internal.h2.H2R2dbcOffsetStore +import com.typesafe.config.Config +import com.typesafe.config.ConfigFactory +import io.r2dbc.spi.R2dbcException +import org.scalatest.wordspec.AnyWordSpecLike + +object H2OffsetStoreTablesSpec { + + // independent of the dialect of the other tests, the H2 database runs in the same process + val config: Config = ConfigFactory + .parseString(s""" + h2-spec.connection-factory = $${pekko.persistence.r2dbc.connection-factory} + h2-spec.connection-factory { + driver = "h2" + h2.database = "h2-offset-store-tables-spec-${UUID.randomUUID()}" + } + h2-url-spec.connection-factory = $${pekko.persistence.r2dbc.connection-factory} + h2-url-spec.connection-factory { + driver = "h2" + h2.url = "mem:h2-offset-store-tables-url-spec-${UUID.randomUUID()};DB_CLOSE_DELAY=-1" + } + """) + .withFallback(ConfigFactory.load()) + .resolve() +} + +class H2OffsetStoreTablesSpec + extends ScalaTestWithActorTestKit(H2OffsetStoreTablesSpec.config) + with AnyWordSpecLike + with LogCapturing { + + private def settings(overrides: String): R2dbcProjectionSettings = + R2dbcProjectionSettings( + ConfigFactory + .parseString(overrides) + .withFallback(ConfigFactory.parseString(""" + dialect = "h2" + use-connection-factory = "h2-spec.connection-factory" + """)) + .withFallback(system.settings.config.getConfig(R2dbcProjectionSettings.DefaultConfigPath))) + + private def createOffsetStore(projectionId: ProjectionId, settings: R2dbcProjectionSettings): R2dbcOffsetStore = { + val connectionFactory = R2dbcProjectionImpl.connectionFactory(settings, ConfigFactory.empty()) + val r2dbcExecutor = R2dbcProjectionImpl.createR2dbcExecutor(connectionFactory, settings, ConfigFactory.empty()) + R2dbcProjectionImpl.createOffsetStore(projectionId, None, settings, r2dbcExecutor) + } + + private def tablesInSchema(schema: String): Set[String] = { + // the plain connection factory, which doesn't create tables + val connectionFactory = ConnectionFactoryProvider(system).connectionFactoryFor("h2-spec.connection-factory") + val executor = R2dbcProjectionImpl.createR2dbcExecutor(connectionFactory, settings(""), ConfigFactory.empty()) + executor + .select("tables in schema")( + _.createStatement("SELECT table_name FROM information_schema.tables WHERE table_schema = ?") + .bind(0, schema.toUpperCase), + row => row.get("table_name", classOf[String]).toLowerCase) + .futureValue + .toSet + } + + private def genProjectionId() = ProjectionId(UUID.randomUUID().toString, "00") + + "The H2 offset store" must { + + "be created for the h2 dialect" in { + createOffsetStore(genProjectionId(), settings("")) shouldBe a[H2R2dbcOffsetStore] + settings("").dialect shouldBe Dialect.H2 + } + + "create the tables of its own schema and table names" in { + val projectionSettings = settings(""" + offset-store { + schema = "h2_spec_custom" + offset-table = "custom_offset" + timestamp-offset-table = "custom_timestamp_offset" + management-table = "custom_management" + } + """) + val projectionId = genProjectionId() + val offsetStore = createOffsetStore(projectionId, projectionSettings) + + offsetStore.saveOffset(OffsetPidSeqNr(1L)).futureValue + offsetStore.readOffset[Long]().futureValue shouldBe Some(1L) + + tablesInSchema("h2_spec_custom") shouldBe Set("custom_offset", "custom_timestamp_offset", "custom_management") + } + + "update offsets and management state with MERGE" in { + val offsetStore = createOffsetStore(genProjectionId(), settings("")) + + offsetStore.saveOffset(OffsetPidSeqNr(1L)).futureValue + offsetStore.saveOffset(OffsetPidSeqNr(2L)).futureValue + offsetStore.readOffset[Long]().futureValue shouldBe Some(2L) + + offsetStore.savePaused(paused = true).futureValue + offsetStore.readManagementState().futureValue shouldBe Some(ManagementState(paused = true)) + offsetStore.savePaused(paused = false).futureValue + offsetStore.readManagementState().futureValue shouldBe Some(ManagementState(paused = false)) + } + + "not create the offset table when it is not used" in { + val projectionSettings = settings(""" + offset-store { + schema = "h2_spec_no_offset_table" + offset-table = "" + } + """) + val offsetStore = createOffsetStore(genProjectionId(), projectionSettings) + + offsetStore.savePaused(paused = true).futureValue + + tablesInSchema("h2_spec_no_offset_table") shouldBe Set( + projectionSettings.timestampOffsetTable, + projectionSettings.managementTable) + } + + "not create the tables when the connection is configured with a url" in { + val offsetStore = createOffsetStore(genProjectionId(), + settings(""" + use-connection-factory = "h2-url-spec.connection-factory" + """)) + + val exc = offsetStore.readManagementState().failed.futureValue + exc shouldBe a[R2dbcException] + withClue(exc.toString) { + TestDbLifecycle.isH2TableNotFound(exc.asInstanceOf[R2dbcException]) shouldBe true + } + } + } +} diff --git a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/R2dbcProjectionSpec.scala b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/R2dbcProjectionSpec.scala index 7c765ab9..b54bcd12 100644 --- a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/R2dbcProjectionSpec.scala +++ b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/R2dbcProjectionSpec.scala @@ -98,8 +98,9 @@ object R2dbcProjectionSpec { def createTableSql(dialect: Dialect): String = { val quotedTable = dialect match { - case Dialect.MySQL => table - case _ => s""""$table"""" + // H2 would make an unquoted name upper case, so it must not be quoted in one place only + case Dialect.MySQL | Dialect.H2 => table + case _ => s""""$table"""" } s"""|CREATE table IF NOT EXISTS $quotedTable ( | id VARCHAR(255) NOT NULL, @@ -157,6 +158,10 @@ object R2dbcProjectionSpec { id = excluded.id, concatenated = excluded.concatenated """ + case Dialect.H2 => + sql""" + MERGE INTO $table (id, concatenated) KEY(id) VALUES (?, ?) + """ } val stmt = session .createStatement(stmtSql) diff --git a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/RuntimePluginConfigSpec.scala b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/RuntimePluginConfigSpec.scala index 3fe20557..fcc92acc 100644 --- a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/RuntimePluginConfigSpec.scala +++ b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/RuntimePluginConfigSpec.scala @@ -22,9 +22,7 @@ package org.apache.pekko.projection.r2dbc import java.util.UUID import scala.collection.immutable.ListSet -import scala.concurrent.Await import scala.concurrent.Future -import scala.concurrent.duration.DurationInt import org.apache.pekko import pekko.Done @@ -36,6 +34,7 @@ import pekko.actor.typed.Behavior import pekko.persistence.Persistence import pekko.persistence.query.typed.EventEnvelope import pekko.persistence.r2dbc.ConnectionFactoryProvider +import pekko.persistence.r2dbc.ConnectionFactorySettings import pekko.persistence.r2dbc.JournalSettings import pekko.persistence.r2dbc.SnapshotSettings import pekko.persistence.r2dbc.internal.R2dbcExecutor @@ -75,6 +74,8 @@ object RuntimePluginConfigSpec { $configKey = { connection-factory { database = "$database" + # one in-memory database per plugin, like the separate databases of the other dialects + h2.database = "$database" } journal.$configKey.connection-factory = $${$configKey.connection-factory} @@ -157,32 +158,20 @@ class RuntimePluginConfigSpec extends ScalaTestWithActorTestKit(TestConfig.confi new R2dbcExecutor( connectionFactoryProvider, LoggerFactory.getLogger(getClass), - journalSettings.logDbCallsExceeding)(system.executionContext, system) + journalSettings.logDbCallsExceeding, + ConnectionFactorySettings.closeCallsExceeding(journalConfig.getConfig(journalSettings.useConnectionFactory)))( + system.executionContext, + system) - Await.result( - r2dbcExecutor.updateOne("beforeAll delete")( - _.createStatement(s"delete from ${journalSettings.journalTableWithSchema}")), - 10.seconds) - Await.result( - r2dbcExecutor.updateOne("beforeAll delete")( - _.createStatement(s"delete from ${snapshotSettings.snapshotsTableWithSchema}")), - 10.seconds) + val dialect = journalSettings.dialect + TestDbLifecycle.deleteAll(r2dbcExecutor, dialect, journalSettings.journalTableWithSchema) + TestDbLifecycle.deleteAll(r2dbcExecutor, dialect, snapshotSettings.snapshotsTableWithSchema) val r2dbcProjectionSettings = R2dbcProjectionSettings(eventSourced.projectionConfig, system) - if (r2dbcProjectionSettings.isOffsetTableDefined) { - Await.result( - r2dbcExecutor.updateOne("beforeAll delete")( - _.createStatement(s"delete from ${r2dbcProjectionSettings.offsetTableWithSchema}")), - 10.seconds) - } - Await.result( - r2dbcExecutor.updateOne("beforeAll delete")( - _.createStatement(s"delete from ${r2dbcProjectionSettings.timestampOffsetTableWithSchema}")), - 10.seconds) - Await.result( - r2dbcExecutor.updateOne("beforeAll delete")( - _.createStatement(s"delete from ${r2dbcProjectionSettings.managementTableWithSchema}")), - 10.seconds) + if (r2dbcProjectionSettings.isOffsetTableDefined) + TestDbLifecycle.deleteAll(r2dbcExecutor, dialect, r2dbcProjectionSettings.offsetTableWithSchema) + TestDbLifecycle.deleteAll(r2dbcExecutor, dialect, r2dbcProjectionSettings.timestampOffsetTableWithSchema) + TestDbLifecycle.deleteAll(r2dbcExecutor, dialect, r2dbcProjectionSettings.managementTableWithSchema) } } diff --git a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/TestConfig.scala b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/TestConfig.scala index aadf9a42..2760dc8b 100644 --- a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/TestConfig.scala +++ b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/TestConfig.scala @@ -59,6 +59,17 @@ object TestConfig { use-app-timestamp = on } """) + case "h2" => + ConfigFactory.parseString(""" + pekko.persistence.r2dbc { + connection-factory { + driver = "h2" + h2.protocol = "mem" + } + db-timestamp-monotonic-increasing = on + use-app-timestamp = on + } + """) } // using load here so that connection-factory can be overridden diff --git a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/TestDbLifecycle.scala b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/TestDbLifecycle.scala index fd6cf94b..675f2b11 100644 --- a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/TestDbLifecycle.scala +++ b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/TestDbLifecycle.scala @@ -14,18 +14,43 @@ package org.apache.pekko.projection.r2dbc import scala.concurrent.Await +import scala.concurrent.ExecutionContext import scala.concurrent.duration._ import org.apache.pekko import pekko.actor.typed.ActorSystem import pekko.persistence.Persistence -import pekko.persistence.r2dbc.ConnectionFactoryProvider +import pekko.persistence.r2dbc.Dialect import pekko.persistence.r2dbc.JournalSettings import pekko.persistence.r2dbc.StateSettings import pekko.persistence.r2dbc.internal.R2dbcExecutor +import pekko.projection.r2dbc.internal.R2dbcProjectionImpl +import com.typesafe.config.ConfigFactory +import io.r2dbc.spi.R2dbcException import org.scalatest.BeforeAndAfterAll import org.scalatest.Suite -import org.slf4j.LoggerFactory + +object TestDbLifecycle { + + // H2 error codes of a table that doesn't exist: TABLE_OR_VIEW_NOT_FOUND_1, _WITH_CANDIDATES_2 and _DATABASE_EMPTY_1 + def isH2TableNotFound(e: R2dbcException): Boolean = + Set(42102, 42103, 42104).contains(e.getErrorCode) + + /** + * Deletes all rows of `table`. The H2 persistence plugins create their tables when they are first used, which can + * be after a test harness deletes from them, so with H2 a table that doesn't exist yet counts as empty. + */ + def deleteAll(r2dbcExecutor: R2dbcExecutor, dialect: Dialect, table: String): Unit = { + val deleted = r2dbcExecutor.updateOne("beforeAll delete")(_.createStatement(s"delete from $table")) + val result = + if (dialect == Dialect.H2) + deleted.recover { + case e: R2dbcException if isH2TableNotFound(e) => 0L + }(ExecutionContext.parasitic) + else deleted + Await.result(result, 10.seconds) + } +} trait TestDbLifecycle extends BeforeAndAfterAll { this: Suite => @@ -36,12 +61,12 @@ trait TestDbLifecycle extends BeforeAndAfterAll { this: Suite => lazy val r2dbcProjectionSettings: R2dbcProjectionSettings = R2dbcProjectionSettings(typedSystem.settings.config.getConfig(testConfigPath)) - lazy val r2dbcExecutor: R2dbcExecutor = { - new R2dbcExecutor( - ConnectionFactoryProvider(typedSystem).connectionFactoryFor(r2dbcProjectionSettings.useConnectionFactory), - LoggerFactory.getLogger(getClass), - r2dbcProjectionSettings.logDbCallsExceeding)(typedSystem.executionContext, typedSystem) - } + // same connection factory and executor as the projections, which creates the offset store tables with H2 + lazy val r2dbcExecutor: R2dbcExecutor = + R2dbcProjectionImpl.createR2dbcExecutor( + R2dbcProjectionImpl.connectionFactory(r2dbcProjectionSettings, ConfigFactory.empty())(typedSystem), + r2dbcProjectionSettings, + ConfigFactory.empty())(typedSystem) lazy val persistenceExt: Persistence = Persistence(typedSystem) @@ -50,28 +75,13 @@ trait TestDbLifecycle extends BeforeAndAfterAll { this: Suite => new JournalSettings(typedSystem.settings.config.getConfig("pekko.persistence.r2dbc.journal")) val stateSettings: StateSettings = new StateSettings(typedSystem.settings.config.getConfig("pekko.persistence.r2dbc.state")) - Await.result( - r2dbcExecutor.updateOne("beforeAll delete")( - _.createStatement(s"delete from ${journalSettings.journalTableWithSchema}")), - 10.seconds) - Await.result( - r2dbcExecutor.updateOne("beforeAll delete")( - _.createStatement(s"delete from ${stateSettings.durableStateTableWithSchema}")), - 10.seconds) - if (r2dbcProjectionSettings.isOffsetTableDefined) { - Await.result( - r2dbcExecutor.updateOne("beforeAll delete")( - _.createStatement(s"delete from ${r2dbcProjectionSettings.offsetTableWithSchema}")), - 10.seconds) - } - Await.result( - r2dbcExecutor.updateOne("beforeAll delete")( - _.createStatement(s"delete from ${r2dbcProjectionSettings.timestampOffsetTableWithSchema}")), - 10.seconds) - Await.result( - r2dbcExecutor.updateOne("beforeAll delete")( - _.createStatement(s"delete from ${r2dbcProjectionSettings.managementTableWithSchema}")), - 10.seconds) + val dialect = r2dbcProjectionSettings.dialect + TestDbLifecycle.deleteAll(r2dbcExecutor, dialect, journalSettings.journalTableWithSchema) + TestDbLifecycle.deleteAll(r2dbcExecutor, dialect, stateSettings.durableStateTableWithSchema) + if (r2dbcProjectionSettings.isOffsetTableDefined) + TestDbLifecycle.deleteAll(r2dbcExecutor, dialect, r2dbcProjectionSettings.offsetTableWithSchema) + TestDbLifecycle.deleteAll(r2dbcExecutor, dialect, r2dbcProjectionSettings.timestampOffsetTableWithSchema) + TestDbLifecycle.deleteAll(r2dbcExecutor, dialect, r2dbcProjectionSettings.managementTableWithSchema) super.beforeAll() } diff --git a/r2dbc/src/main/resources/reference.conf b/r2dbc/src/main/resources/reference.conf index 7ca39ced..1aafbfc2 100644 --- a/r2dbc/src/main/resources/reference.conf +++ b/r2dbc/src/main/resources/reference.conf @@ -5,7 +5,7 @@ //#projection-config pekko.projection.r2dbc { - # postgres, yugabyte or mysql + # postgres, yugabyte, mysql or h2 dialect = ${pekko.persistence.r2dbc.dialect} offset-store { diff --git a/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/R2dbcOffsetStore.scala b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/R2dbcOffsetStore.scala index 9ce55b0c..bdcbfd20 100644 --- a/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/R2dbcOffsetStore.scala +++ b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/R2dbcOffsetStore.scala @@ -50,6 +50,7 @@ import pekko.projection.internal.OffsetSerialization import pekko.projection.internal.OffsetSerialization.MultipleOffsets import pekko.projection.internal.OffsetSerialization.SingleOffset import pekko.projection.r2dbc.R2dbcProjectionSettings +import pekko.projection.r2dbc.internal.h2.H2R2dbcOffsetStore import pekko.projection.r2dbc.internal.mysql.MySQLR2dbcOffsetStore import io.r2dbc.spi.Connection import io.r2dbc.spi.Statement @@ -191,6 +192,8 @@ private[projection] object R2dbcOffsetStore { new R2dbcOffsetStore(projectionId, sourceProvider, system, settings, r2dbcExecutor, clock) case Dialect.MySQL => new MySQLR2dbcOffsetStore(projectionId, sourceProvider, system, settings, r2dbcExecutor, clock) + case Dialect.H2 => + new H2R2dbcOffsetStore(projectionId, sourceProvider, system, settings, r2dbcExecutor, clock) } final class RejectedEnvelope(message: String) extends IllegalStateException(message) diff --git a/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/R2dbcProjectionImpl.scala b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/R2dbcProjectionImpl.scala index 41f10257..2cdacb73 100644 --- a/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/R2dbcProjectionImpl.scala +++ b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/R2dbcProjectionImpl.scala @@ -32,6 +32,9 @@ import pekko.persistence.query.DeletedDurableState import pekko.persistence.query.UpdatedDurableState import pekko.persistence.query.typed.EventEnvelope import pekko.persistence.query.typed.scaladsl.LoadEventQuery +import pekko.persistence.r2dbc.ConnectionFactoryProvider +import pekko.persistence.r2dbc.ConnectionFactorySettings +import pekko.persistence.r2dbc.Dialect import pekko.persistence.r2dbc.internal.R2dbcExecutor import pekko.persistence.state.scaladsl.DurableStateStore import pekko.persistence.state.scaladsl.GetObjectResult @@ -64,6 +67,7 @@ import pekko.projection.javadsl import pekko.projection.r2dbc.R2dbcProjectionSettings import pekko.projection.r2dbc.internal.R2dbcOffsetStore.RejectedEnvelope import pekko.projection.r2dbc.internal.R2dbcProjectionImpl.extractOffsetPidSeqNr +import pekko.projection.r2dbc.internal.h2.H2OffsetStoreTables import pekko.projection.r2dbc.scaladsl.R2dbcHandler import pekko.projection.r2dbc.scaladsl.R2dbcSession import pekko.projection.scaladsl @@ -72,6 +76,7 @@ import pekko.projection.scaladsl.SourceProvider import pekko.stream.RestartSettings import pekko.stream.scaladsl.FlowWithContext import pekko.stream.scaladsl.Source +import com.typesafe.config.Config import io.r2dbc.spi.ConnectionFactory import org.slf4j.Logger import org.slf4j.LoggerFactory @@ -92,12 +97,48 @@ private[projection] object R2dbcProjectionImpl { projectionId: ProjectionId, sourceProvider: Option[BySlicesSourceProvider], settings: R2dbcProjectionSettings, - connectionFactory: ConnectionFactory)(implicit system: ActorSystem[?]) = { - val r2dbcExecutor = - new R2dbcExecutor(connectionFactory, log, settings.logDbCallsExceeding)(system.executionContext, system) + r2dbcExecutor: R2dbcExecutor)(implicit system: ActorSystem[?]): R2dbcOffsetStore = R2dbcOffsetStore.fromConfig(projectionId, sourceProvider, system, settings, r2dbcExecutor) + + /** + * The connection factory of `settings.useConnectionFactory`. With the H2 dialect it creates the offset store + * tables before the first connection is handed out, since there is no schema script for an in-process database. + */ + private[projection] def connectionFactory(settings: R2dbcProjectionSettings, config: Config)( + implicit system: ActorSystem[?]): ConnectionFactory = { + val connectionFactory = + ConnectionFactoryProvider(system).connectionFactoryFor(settings.useConnectionFactory, config) + settings.dialect match { + case Dialect.H2 => + H2OffsetStoreTables.creatingTables(connectionFactory, connectionFactoryConfig(settings, config), settings) + case _ => connectionFactory + } } + /** + * With the H2 dialect the database calls run on the `h2.use-dispatcher` of the connection factory, because the + * H2 driver blocks. + */ + private[projection] def createR2dbcExecutor( + connectionFactory: ConnectionFactory, + settings: R2dbcProjectionSettings, + config: Config)(implicit system: ActorSystem[?]): R2dbcExecutor = { + val cfConfig = connectionFactoryConfig(settings, config) + val executionContext = settings.dialect match { + case Dialect.H2 => H2OffsetStoreTables.executionContext(system, cfConfig) + case _ => system.executionContext + } + new R2dbcExecutor( + connectionFactory, + log, + settings.logDbCallsExceeding, + ConnectionFactorySettings.closeCallsExceeding(cfConfig))(executionContext, system) + } + + private def connectionFactoryConfig(settings: R2dbcProjectionSettings, config: Config)( + implicit system: ActorSystem[?]): Config = + config.withFallback(system.settings.config).getConfig(settings.useConnectionFactory) + private val loadEnvelopeCounter = new AtomicLong def loadEnvelope[Envelope](env: Envelope, sourceProvider: SourceProvider[?, Envelope])( diff --git a/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/h2/H2OffsetStoreTables.scala b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/h2/H2OffsetStoreTables.scala new file mode 100644 index 00000000..69dab89b --- /dev/null +++ b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/h2/H2OffsetStoreTables.scala @@ -0,0 +1,133 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * license agreements; and to You under the Apache License, version 2.0: + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * This file is part of the Apache Pekko project, which was derived from Akka. + */ + +/* + * Copyright (C) 2022 - 2023 Lightbend Inc. + */ + +package org.apache.pekko.projection.r2dbc.internal.h2 + +import java.time.Duration + +import scala.concurrent.ExecutionContext +import scala.jdk.CollectionConverters._ + +import org.apache.pekko +import pekko.actor.typed.ActorSystem +import pekko.actor.typed.DispatcherSelector +import pekko.annotation.InternalApi +import pekko.projection.r2dbc.R2dbcProjectionSettings +import com.typesafe.config.Config +import io.r2dbc.spi.Connection +import io.r2dbc.spi.ConnectionFactory +import io.r2dbc.spi.ConnectionFactoryMetadata +import io.r2dbc.spi.Result +import org.reactivestreams.Publisher +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono + +/** + * INTERNAL API + * + * The H2 database runs in the same process, so its tables can't be created up front with a schema script. Like the + * pekko-persistence-r2dbc plugins with the h2 dialect, the offset store creates the tables of its own settings before + * the first connection is handed out. + */ +@InternalApi +private[projection] object H2OffsetStoreTables { + + /** + * Wraps the connection factory so that the offset store tables of `settings` are created before the first + * connection is handed out. Not when the connection is configured with a `url`, same as the persistence plugins. + */ + def creatingTables( + connectionFactory: ConnectionFactory, + connectionFactoryConfig: Config, + settings: R2dbcProjectionSettings): ConnectionFactory = + if (connectionFactoryConfig.getString("h2.url").nonEmpty) connectionFactory + else new TableCreatingConnectionFactory(connectionFactory, ddl(settings)) + + /** + * The H2 R2DBC driver blocks in surprising places, so the database calls run on the `h2.use-dispatcher` of the + * connection factory, same as the persistence plugins. + */ + def executionContext(system: ActorSystem[?], connectionFactoryConfig: Config): ExecutionContext = + system.dispatchers.lookup(DispatcherSelector.fromConfig(connectionFactoryConfig.getString("h2.use-dispatcher"))) + + def ddl(settings: R2dbcProjectionSettings): Seq[String] = { + val schema = settings.schema.map(s => s"CREATE SCHEMA IF NOT EXISTS $s").toSeq + + val offsetTable = + if (settings.isOffsetTableDefined) + Seq(s"""CREATE TABLE IF NOT EXISTS ${settings.offsetTableWithSchema} ( + projection_name VARCHAR(255) NOT NULL, + projection_key VARCHAR(255) NOT NULL, + current_offset VARCHAR(255) NOT NULL, + manifest VARCHAR(32) NOT NULL, + mergeable BOOLEAN NOT NULL, + last_updated BIGINT NOT NULL, + PRIMARY KEY(projection_name, projection_key) + )""") + else Seq.empty + + val timestampOffsetTable = + s"""CREATE TABLE IF NOT EXISTS ${settings.timestampOffsetTableWithSchema} ( + projection_name VARCHAR(255) NOT NULL, + projection_key VARCHAR(255) NOT NULL, + slice INT NOT NULL, + persistence_id VARCHAR(255) NOT NULL, + seq_nr BIGINT NOT NULL, + timestamp_offset timestamp with time zone NOT NULL, + timestamp_consumed timestamp with time zone NOT NULL, + PRIMARY KEY(slice, projection_name, timestamp_offset, persistence_id, seq_nr) + )""" + + val managementTable = + s"""CREATE TABLE IF NOT EXISTS ${settings.managementTableWithSchema} ( + 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) + )""" + + schema ++ offsetTable ++ Seq(timestampOffsetTable, managementTable) + } + + /** + * Runs `ddl` once, on a connection of its own, before handing out the first connection. A failure is not cached, + * so the next `create()` runs `ddl` again, for example after a transient lock on a file database. + */ + private final class TableCreatingConnectionFactory(underlying: ConnectionFactory, ddl: Seq[String]) + extends ConnectionFactory { + + private val tablesCreated: Mono[Void] = + Mono + .usingWhen[Void, Connection]( + Mono.from[Connection](underlying.create()), + (connection: Connection) => + Flux + .fromIterable(ddl.asJava) + .concatMap((statement: String) => + Flux + .from[Result](connection.createStatement(statement).execute()) + .concatMap((result: Result) => result.getRowsUpdated)) + .`then`(), + (connection: Connection) => connection.close()) + .cache( + (_: Void) => Duration.ofMillis(Long.MaxValue), + (_: Throwable) => Duration.ZERO, + () => Duration.ofMillis(Long.MaxValue)) + + override def create(): Publisher[? <: Connection] = + tablesCreated.`then`(Mono.from[Connection](underlying.create())) + + override def getMetadata: ConnectionFactoryMetadata = underlying.getMetadata + } +} diff --git a/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/h2/H2R2dbcOffsetStore.scala b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/h2/H2R2dbcOffsetStore.scala new file mode 100644 index 00000000..442c250c --- /dev/null +++ b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/h2/H2R2dbcOffsetStore.scala @@ -0,0 +1,55 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * license agreements; and to You under the Apache License, version 2.0: + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * This file is part of the Apache Pekko project, which was derived from Akka. + */ + +/* + * Copyright (C) 2023 Lightbend Inc. + */ + +package org.apache.pekko.projection.r2dbc.internal.h2 + +import java.time.Clock + +import org.apache.pekko +import pekko.actor.typed.ActorSystem +import pekko.annotation.InternalApi +import pekko.persistence.r2dbc.internal.R2dbcExecutor +import pekko.persistence.r2dbc.internal.Sql.DialectInterpolation +import pekko.projection.BySlicesSourceProvider +import pekko.projection.ProjectionId +import pekko.projection.r2dbc.R2dbcProjectionSettings +import pekko.projection.r2dbc.internal.R2dbcOffsetStore + +/** + * INTERNAL API + */ +@InternalApi +private[projection] class H2R2dbcOffsetStore( + projectionId: ProjectionId, + sourceProvider: Option[BySlicesSourceProvider], + system: ActorSystem[?], + settings: R2dbcProjectionSettings, + r2dbcExecutor: R2dbcExecutor, + clock: Clock = Clock.systemUTC()) + extends R2dbcOffsetStore(projectionId, sourceProvider, system, settings, r2dbcExecutor, clock) { + + // H2 has no transaction_timestamp(), CURRENT_TIMESTAMP is the start of the transaction + override lazy val timestampSql: String = "CURRENT_TIMESTAMP" + + override val upsertOffsetSql: String = sql""" + MERGE INTO $offsetTable + (projection_name, projection_key, current_offset, manifest, mergeable, last_updated) + KEY(projection_name, projection_key) + VALUES (?,?,?,?,?,?)""" + + override val updateManagementStateSql: String = sql""" + MERGE INTO $managementTable + (projection_name, projection_key, paused, last_updated) + KEY(projection_name, projection_key) + VALUES (?,?,?,?)""" +} diff --git a/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/scaladsl/R2dbcProjection.scala b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/scaladsl/R2dbcProjection.scala index 9872a938..6441f921 100644 --- a/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/scaladsl/R2dbcProjection.scala +++ b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/scaladsl/R2dbcProjection.scala @@ -17,8 +17,6 @@ import org.apache.pekko import pekko.Done import pekko.actor.typed.ActorSystem import pekko.annotation.ApiMayChange -import pekko.persistence.r2dbc.ConnectionFactoryProvider -import pekko.persistence.r2dbc.internal.R2dbcExecutor import pekko.projection.BySlicesSourceProvider import pekko.projection.ProjectionContext import pekko.projection.ProjectionId @@ -40,7 +38,6 @@ import pekko.projection.scaladsl.SourceProvider import pekko.stream.scaladsl.FlowWithContext import com.typesafe.config.Config import com.typesafe.config.ConfigFactory -import io.r2dbc.spi.ConnectionFactory @ApiMayChange object R2dbcProjection { @@ -68,16 +65,14 @@ object R2dbcProjection { handler: () => R2dbcHandler[Envelope])(implicit system: ActorSystem[?]): ExactlyOnceProjection[Offset, Envelope] = { val r2dbcSettings = settings.getOrElse(R2dbcProjectionSettings(config, system)) - val connFactory = connectionFactory(system, config, r2dbcSettings) + val connFactory = R2dbcProjectionImpl.connectionFactory(r2dbcSettings, config) + val r2dbcExecutor = R2dbcProjectionImpl.createR2dbcExecutor(connFactory, r2dbcSettings, config) val offsetStore = R2dbcProjectionImpl.createOffsetStore( projectionId, timestampOffsetBySlicesSourceProvider(sourceProvider), r2dbcSettings, - connFactory) - val r2dbcExecutor = new R2dbcExecutor(connFactory, R2dbcProjectionImpl.log, r2dbcSettings.logDbCallsExceeding)( - system.executionContext, - system) + r2dbcExecutor) val adaptedHandler = R2dbcProjectionImpl.adaptedHandlerForExactlyOnce(sourceProvider, handler, offsetStore, r2dbcExecutor)( @@ -128,16 +123,14 @@ object R2dbcProjection { system: ActorSystem[?]): AtLeastOnceProjection[Offset, Envelope] = { val r2dbcSettings = settings.getOrElse(R2dbcProjectionSettings(config, system)) - val connFactory = connectionFactory(system, config, r2dbcSettings) + val connFactory = R2dbcProjectionImpl.connectionFactory(r2dbcSettings, config) + val r2dbcExecutor = R2dbcProjectionImpl.createR2dbcExecutor(connFactory, r2dbcSettings, config) val offsetStore = R2dbcProjectionImpl.createOffsetStore( projectionId, timestampOffsetBySlicesSourceProvider(sourceProvider), r2dbcSettings, - connFactory) - val r2dbcExecutor = new R2dbcExecutor(connFactory, R2dbcProjectionImpl.log, r2dbcSettings.logDbCallsExceeding)( - system.executionContext, - system) + r2dbcExecutor) val adaptedHandler = R2dbcProjectionImpl.adaptedHandlerForAtLeastOnce(sourceProvider, handler, offsetStore, r2dbcExecutor)( @@ -186,13 +179,13 @@ object R2dbcProjection { handler: () => Handler[Envelope])(implicit system: ActorSystem[?]): AtLeastOnceProjection[Offset, Envelope] = { val r2dbcSettings = settings.getOrElse(R2dbcProjectionSettings(config, system)) - val connFactory = connectionFactory(system, config, r2dbcSettings) + val connFactory = R2dbcProjectionImpl.connectionFactory(r2dbcSettings, config) val offsetStore = R2dbcProjectionImpl.createOffsetStore( projectionId, timestampOffsetBySlicesSourceProvider(sourceProvider), r2dbcSettings, - connFactory) + R2dbcProjectionImpl.createR2dbcExecutor(connFactory, r2dbcSettings, config)) val adaptedHandler = R2dbcProjectionImpl.adaptedHandlerForAtLeastOnceAsync(sourceProvider, handler, offsetStore)( @@ -238,16 +231,14 @@ object R2dbcProjection { system: ActorSystem[?]): GroupedProjection[Offset, Envelope] = { val r2dbcSettings = settings.getOrElse(R2dbcProjectionSettings(config, system)) - val connFactory = connectionFactory(system, config, r2dbcSettings) + val connFactory = R2dbcProjectionImpl.connectionFactory(r2dbcSettings, config) + val r2dbcExecutor = R2dbcProjectionImpl.createR2dbcExecutor(connFactory, r2dbcSettings, config) val offsetStore = R2dbcProjectionImpl.createOffsetStore( projectionId, timestampOffsetBySlicesSourceProvider(sourceProvider), r2dbcSettings, - connFactory) - val r2dbcExecutor = new R2dbcExecutor(connFactory, R2dbcProjectionImpl.log, r2dbcSettings.logDbCallsExceeding)( - system.executionContext, - system) + r2dbcExecutor) val adaptedHandler = R2dbcProjectionImpl.adaptedHandlerForGrouped(sourceProvider, handler, offsetStore, r2dbcExecutor)( @@ -297,13 +288,13 @@ object R2dbcProjection { system: ActorSystem[?]): GroupedProjection[Offset, Envelope] = { val r2dbcSettings = settings.getOrElse(R2dbcProjectionSettings(config, system)) - val connFactory = connectionFactory(system, config, r2dbcSettings) + val connFactory = R2dbcProjectionImpl.connectionFactory(r2dbcSettings, config) val offsetStore = R2dbcProjectionImpl.createOffsetStore( projectionId, timestampOffsetBySlicesSourceProvider(sourceProvider), r2dbcSettings, - connFactory) + R2dbcProjectionImpl.createR2dbcExecutor(connFactory, r2dbcSettings, config)) val adaptedHandler = R2dbcProjectionImpl.adaptedHandlerForGroupedAsync(sourceProvider, handler, offsetStore)( @@ -361,13 +352,13 @@ object R2dbcProjection { system: ActorSystem[?]): AtLeastOnceFlowProjection[Offset, Envelope] = { val r2dbcSettings = settings.getOrElse(R2dbcProjectionSettings(config, system)) - val connFactory = connectionFactory(system, config, r2dbcSettings) + val connFactory = R2dbcProjectionImpl.connectionFactory(r2dbcSettings, config) val offsetStore = R2dbcProjectionImpl.createOffsetStore( projectionId, timestampOffsetBySlicesSourceProvider(sourceProvider), r2dbcSettings, - connFactory) + R2dbcProjectionImpl.createR2dbcExecutor(connFactory, r2dbcSettings, config)) val adaptedHandler = R2dbcProjectionImpl.adaptedHandlerForFlow(sourceProvider, handler, offsetStore, r2dbcSettings)(system) @@ -384,11 +375,6 @@ object R2dbcProjection { offsetStore) } - private def connectionFactory(system: ActorSystem[?], config: Config, r2dbcSettings: R2dbcProjectionSettings) - : ConnectionFactory = { - ConnectionFactoryProvider(system).connectionFactoryFor(r2dbcSettings.useConnectionFactory, config) - } - private def timestampOffsetBySlicesSourceProvider( sourceProvider: SourceProvider[?, ?]): Option[BySlicesSourceProvider] = { sourceProvider match { From 18e14970b790034e07144838dfbb08ff15effd2c Mon Sep 17 00:00:00 2001 From: Domantas Petrauskas Date: Tue, 27 Oct 2026 10:58:01 +0200 Subject: [PATCH 2/3] r2dbc: use the shared executor factory in R2dbcSession Motivation: main gained R2dbcSession.withSession (#664), which creates an R2dbcExecutor with the constructor that the pekko-persistence-r2dbc snapshot deprecates, so the build fails with warnings as errors. The spy offset store from #620 has no H2 case. Modification: R2dbcSession.withSession creates its executor with R2dbcProjectionImpl.createR2dbcExecutor, which takes the logger to keep its own logger name. That also runs H2 sessions on h2.use-dispatcher and reads close-calls-exceeding. Add the H2 case to the spy offset store in R2dbcTimestampOffsetProjectionSpec. Result: The branch compiles and passes with the changes from main. Tests: - r2dbc-int-test/test with h2 dialect: 136 succeeded, 0 failed - r2dbc-int-test/test with PostgreSQL: 136 succeeded, 0 failed - Test/compile, checkCodeStyle; mimaReportBinaryIssues, +headerCheckAll, docs/paradox, ++3.3.8 r2dbc-int-test/Test/compile: pass References: Refs #662 --- .../r2dbc/R2dbcTimestampOffsetProjectionSpec.scala | 8 ++++++++ .../r2dbc/internal/R2dbcProjectionImpl.scala | 5 +++-- .../projection/r2dbc/scaladsl/R2dbcSession.scala | 11 +++++++---- 3 files changed, 18 insertions(+), 6 deletions(-) diff --git a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/R2dbcTimestampOffsetProjectionSpec.scala b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/R2dbcTimestampOffsetProjectionSpec.scala index 92ba4c94..ca666910 100644 --- a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/R2dbcTimestampOffsetProjectionSpec.scala +++ b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/R2dbcTimestampOffsetProjectionSpec.scala @@ -53,6 +53,7 @@ import pekko.projection.internal.NoopStatusObserver import pekko.projection.internal.SingleHandlerStrategy import pekko.projection.r2dbc.internal.R2dbcOffsetStore import pekko.projection.r2dbc.internal.R2dbcProjectionImpl +import pekko.projection.r2dbc.internal.h2.H2R2dbcOffsetStore import pekko.projection.r2dbc.internal.mysql.MySQLR2dbcOffsetStore import pekko.projection.r2dbc.scaladsl.R2dbcHandler import pekko.projection.r2dbc.scaladsl.R2dbcProjection @@ -1479,6 +1480,13 @@ class R2dbcTimestampOffsetProjectionSpec super.deleteOldTimestampOffsets() } } + case Dialect.H2 => + new H2R2dbcOffsetStore(projectionId, Some(sourceProvider), system, spySettings, r2dbcExecutor) { + override def deleteOldTimestampOffsets(): Future[Long] = { + deleteCount.incrementAndGet() + super.deleteOldTimestampOffsets() + } + } } } diff --git a/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/R2dbcProjectionImpl.scala b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/R2dbcProjectionImpl.scala index b00c50e9..e02d6535 100644 --- a/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/R2dbcProjectionImpl.scala +++ b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/R2dbcProjectionImpl.scala @@ -122,7 +122,8 @@ private[projection] object R2dbcProjectionImpl { private[projection] def createR2dbcExecutor( connectionFactory: ConnectionFactory, settings: R2dbcProjectionSettings, - config: Config)(implicit system: ActorSystem[?]): R2dbcExecutor = { + config: Config, + logger: Logger = log)(implicit system: ActorSystem[?]): R2dbcExecutor = { val cfConfig = connectionFactoryConfig(settings, config) val executionContext = settings.dialect match { case Dialect.H2 => H2OffsetStoreTables.executionContext(system, cfConfig) @@ -130,7 +131,7 @@ private[projection] object R2dbcProjectionImpl { } new R2dbcExecutor( connectionFactory, - log, + logger, settings.logDbCallsExceeding, ConnectionFactorySettings.closeCallsExceeding(cfConfig))(executionContext, system) } diff --git a/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/scaladsl/R2dbcSession.scala b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/scaladsl/R2dbcSession.scala index 398baa92..5bbd4547 100644 --- a/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/scaladsl/R2dbcSession.scala +++ b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/scaladsl/R2dbcSession.scala @@ -24,6 +24,8 @@ import pekko.annotation.ApiMayChange import pekko.persistence.r2dbc.ConnectionFactoryProvider import pekko.persistence.r2dbc.internal.R2dbcExecutor import pekko.projection.r2dbc.R2dbcProjectionSettings +import pekko.projection.r2dbc.internal.R2dbcProjectionImpl +import com.typesafe.config.ConfigFactory import io.r2dbc.spi.Connection import io.r2dbc.spi.Row import io.r2dbc.spi.Statement @@ -61,11 +63,12 @@ object R2dbcSession { connectionFactoryConfigPath: Option[String], fun: R2dbcSession => Future[A]): Future[A] = try { - val settings = R2dbcProjectionSettings(system) - val connectionFactory = ConnectionFactoryProvider(system).connectionFactoryFor( - connectionFactoryConfigPath.getOrElse(settings.useConnectionFactory)) + val defaultSettings = R2dbcProjectionSettings(system) + val settings = + connectionFactoryConfigPath.fold(defaultSettings)(path => defaultSettings.withUseConnectionFactory(path)) + val connectionFactory = ConnectionFactoryProvider(system).connectionFactoryFor(settings.useConnectionFactory) val r2dbcExecutor = - new R2dbcExecutor(connectionFactory, log, settings.logDbCallsExceeding)(system.executionContext, system) + R2dbcProjectionImpl.createR2dbcExecutor(connectionFactory, settings, ConfigFactory.empty(), log)(system) r2dbcExecutor.withConnection("R2dbcSession") { connection => fun(new R2dbcSession(connection)(system.executionContext, system)) } From 8207a7f450f6c0654da54722fe75e0ac900c29d4 Mon Sep 17 00:00:00 2001 From: Domantas Petrauskas Date: Sat, 31 Oct 2026 12:33:26 +0200 Subject: [PATCH 3/3] r2dbc: support MariaDB Motivation: apache/pekko-persistence-r2dbc#534 added the mariadb dialect to the persistence plugin. R2DBC projections fail on it: the dialect has no offset store, and MariaDB doesn't support the `AS excluded` row alias that the MySQL offset store upserts with. Modification: Add MariaDBR2dbcOffsetStore, the MySQL offset store with VALUES(col) upserts, and select it for the mariadb dialect. Instants need no handling in the projection, as the connection factory of the plugin converts them for the MariaDB driver. Build against the pekko-persistence-r2dbc snapshot that includes apache/pekko-persistence-r2dbc#534, add the MariaDB driver as a test dependency, a MariaDB CI job and docs. Result: R2DBC projections can store offsets in MariaDB with the mariadb dialect. Tests: - MariaDBR2dbcOffsetStoreSpec on MariaDB 12.3.3 (America/New_York): 3 passed; 2 of 3 fail with the MySQL upserts - r2dbc-int-test/test with PostgreSQL and H2: 137 passed, 2 canceled (MariaDB only) - r2dbc-int-test/test with MariaDB and MySQL: event sourced specs fail locally under load on Windows, also on MySQL and the H2 branch - checkCodeStyle, mimaReportBinaryIssues, headerCheckAll, docs/paradox, Scala 3.3.8 and 3.9.0 compile: pass References: Refs apache/pekko-persistence-r2dbc#534 --- .github/workflows/integration-tests-r2dbc.yml | 63 +++++++++ docker-files/docker-compose-mariadb.yml | 32 +++++ docs/src/main/paradox/r2dbc.md | 13 ++ project/Dependencies.scala | 4 + project/PekkoPersistenceR2DBCDependency.scala | 2 +- .../r2dbc/MariaDBR2dbcOffsetStoreSpec.scala | 128 ++++++++++++++++++ .../r2dbc/R2dbcProjectionSpec.scala | 11 +- .../R2dbcTimestampOffsetProjectionSpec.scala | 8 ++ .../pekko/projection/r2dbc/TestConfig.scala | 15 ++ r2dbc/src/main/resources/reference.conf | 4 +- .../r2dbc/internal/R2dbcOffsetStore.scala | 3 + .../mariadb/MariaDBR2dbcOffsetStore.scala | 66 +++++++++ 12 files changed, 344 insertions(+), 5 deletions(-) create mode 100644 docker-files/docker-compose-mariadb.yml create mode 100644 r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/MariaDBR2dbcOffsetStoreSpec.scala create mode 100644 r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/mariadb/MariaDBR2dbcOffsetStore.scala diff --git a/.github/workflows/integration-tests-r2dbc.yml b/.github/workflows/integration-tests-r2dbc.yml index 17dde9df..6d70e39b 100644 --- a/.github/workflows/integration-tests-r2dbc.yml +++ b/.github/workflows/integration-tests-r2dbc.yml @@ -208,6 +208,69 @@ jobs: if: ${{ failure() }} run: find . -name "*.log" -exec ./scripts/cat-log.sh {} \; + mariadb-test: + name: Build and Test Integration for R2DBC with MariaDB + runs-on: ubuntu-22.04 + if: github.repository == 'apache/pekko-projection' + strategy: + fail-fast: false + matrix: + include: + - { java-version: 17, scala-version: 2.13, sbt-opts: '' } + - { java-version: 17, scala-version: next, sbt-opts: '' } + + steps: + - name: Checkout + uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 + with: + fetch-depth: 0 + fetch-tags: true + + - name: Checkout GitHub merge + if: github.event.pull_request + run: |- + git fetch origin pull/${{ github.event.pull_request.number }}/merge:scratch + git checkout scratch + + - name: Cache Coursier cache + uses: coursier/cache-action@057d08c2f9a2c4e7a4d33df3aae657b7db88b8cc # v8.2.0 + + - name: Setup JDK ${{ matrix.java-version }} + uses: actions/setup-java@b843ebf9b913146b38da2eaa7aeddf76074ae1ad # v6.1.0 + with: + java-version: ${{ matrix.java-version }} + distribution: temurin + + - name: Install sbt + uses: sbt/setup-sbt@ecad189dd245444b9ae0a0cd5922fd938b88906f # v1.5.12 + + - name: Start MariaDB + run: |- + docker compose -f docker-files/docker-compose-mariadb.yml up -d --wait + # loop to wait for the DB to be ready + for i in {1..8}; do + if docker exec -i docker-mariadb-db-1 mariadb \ + -h 127.0.0.1 --user=root --password=root --database=mysql -e 'SELECT 1;' 2>/dev/null; then + break + fi + echo "Retry $i: SQL check failed, waiting before retry..." + sleep $((i * 2)) + done + docker exec -i docker-mariadb-db-1 mariadb -h 127.0.0.1 --user=root --password=root --database=mysql < r2dbc-int-test/ddl-scripts/create_tables_mysql.sql + docker exec -i docker-mariadb-db-1 mariadb -h 127.0.0.1 --user=root --password=root -e 'CREATE SCHEMA database1;' + docker exec -i docker-mariadb-db-1 mariadb -h 127.0.0.1 --user=root --password=root --database=database1 < r2dbc-int-test/ddl-scripts/create_tables_mysql.sql + docker exec -i docker-mariadb-db-1 mariadb -h 127.0.0.1 --user=root --password=root -e 'CREATE SCHEMA database2;' + docker exec -i docker-mariadb-db-1 mariadb -h 127.0.0.1 --user=root --password=root --database=database2 < r2dbc-int-test/ddl-scripts/create_tables_mysql.sql + + - name: test + run: |- + scala_version="$(bash .github/scripts/resolve-scala-version.sh "${{ matrix.scala-version }}")" + sbt -Dpekko.persistence.r2dbc.dialect=mariadb -Dpekko.projection.r2dbc.dialect=mariadb "++${scala_version}!" "r2dbc-int-test/test" ${{ matrix.sbt-opts }} + + - name: Print logs on failure + if: ${{ failure() }} + run: find . -name "*.log" -exec ./scripts/cat-log.sh {} \; + h2-test: name: Build and Test Integration for R2DBC with H2 runs-on: ubuntu-22.04 diff --git a/docker-files/docker-compose-mariadb.yml b/docker-files/docker-compose-mariadb.yml new file mode 100644 index 00000000..244bde5e --- /dev/null +++ b/docker-files/docker-compose-mariadb.yml @@ -0,0 +1,32 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +services: + mariadb-db: + image: mariadb:12.3.3 + container_name: docker-mariadb-db-1 + ports: + - 3306:3306 + environment: + MARIADB_ROOT_PASSWORD: root + # a server time zone with daylight saving time, so that the tests verify that timestamps are stored in UTC + TZ: America/New_York + healthcheck: + test: [ "CMD", "healthcheck.sh", "--connect", "--innodb_initialized" ] + interval: 1s + timeout: 1s + retries: 60 diff --git a/docs/src/main/paradox/r2dbc.md b/docs/src/main/paradox/r2dbc.md index f68f2600..ea3f08b8 100644 --- a/docs/src/main/paradox/r2dbc.md +++ b/docs/src/main/paradox/r2dbc.md @@ -54,9 +54,22 @@ YugaByte MySQL : @@snip [MySQL Schema](/r2dbc-int-test/ddl-scripts/create_tables_mysql.sql) +MariaDB +: @@snip [MySQL Schema](/r2dbc-int-test/ddl-scripts/create_tables_mysql.sql) + H2 : The tables are created automatically, see @ref:[H2](#h2). +### MariaDB + +MariaDB 10.11 or later is supported with the `mariadb` dialect and the `org.mariadb:r2dbc-mariadb` driver, see +@extref:[MariaDB in Apache Pekko Persistence R2DBC](pekko-persistence-r2dbc:getting-started.html#mariadb) for how to +configure the connection and the driver dependency. + +The offset store uses the same tables as MySQL. The MariaDB driver doesn't support `java.time.Instant`, so the +connection factory of Apache Pekko Persistence R2DBC binds and reads timestamps as `LocalDateTime` in UTC, with a UTC +database session. The same applies to the statements of an `R2dbcHandler` that run on the `R2dbcSession`. + ### H2 With the `h2` dialect the database runs in the same JVM process, see diff --git a/project/Dependencies.scala b/project/Dependencies.scala index cd67877b..4c24627b 100644 --- a/project/Dependencies.scala +++ b/project/Dependencies.scala @@ -69,6 +69,8 @@ object Dependencies { val r2dbcPool = "io.r2dbc" % "r2dbc-pool" % "1.0.2.RELEASE" val r2dbcPostgres = "org.postgresql" % "r2dbc-postgresql" % "1.1.3.RELEASE" val r2dbcMysql = "io.asyncer" % "r2dbc-mysql" % "1.4.3" + // the version that pekko-persistence-r2dbc is built against + val r2dbcMariadb = "org.mariadb" % "r2dbc-mariadb" % "1.3.2" // ApacheV2 val r2dbcH2 = "io.r2dbc" % "r2dbc-h2" % "1.1.0.RELEASE" // ApacheV2 // the H2 version that pekko-persistence-r2dbc is built against val h2R2dbc = "com.h2database" % "h2" % "2.5.252" // EPL 1.0 or MPL 2.0 @@ -117,6 +119,7 @@ object Dependencies { val r2dbcPostgres = Compile.r2dbcPostgres % "test" val r2dbcMysql = Compile.r2dbcMysql % "test" + val r2dbcMariadb = Compile.r2dbcMariadb % "test" val r2dbcH2 = Compile.r2dbcH2 % "test" val h2R2dbc = Compile.h2R2dbc % "test" } @@ -281,6 +284,7 @@ object Dependencies { Compile.r2dbcPool, Test.r2dbcPostgres, Test.r2dbcMysql, + Test.r2dbcMariadb, Test.r2dbcH2, Test.h2R2dbc, Test.pekkoSerializationJackson, diff --git a/project/PekkoPersistenceR2DBCDependency.scala b/project/PekkoPersistenceR2DBCDependency.scala index 1a62a6ef..41619ff5 100644 --- a/project/PekkoPersistenceR2DBCDependency.scala +++ b/project/PekkoPersistenceR2DBCDependency.scala @@ -21,5 +21,5 @@ import com.github.pjfanning.pekkobuild.PekkoDependency object PekkoPersistenceR2DBCDependency extends PekkoDependency { override val checkProject: String = "pekko-persistence-r2dbc" override val module: Option[String] = None - override val currentVersion: String = "2.0.0-M1+113-ac55a4d8-SNAPSHOT" + override val currentVersion: String = "2.0.0-M1+120-5a212bab-SNAPSHOT" } diff --git a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/MariaDBR2dbcOffsetStoreSpec.scala b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/MariaDBR2dbcOffsetStoreSpec.scala new file mode 100644 index 00000000..bbfb19e1 --- /dev/null +++ b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/MariaDBR2dbcOffsetStoreSpec.scala @@ -0,0 +1,128 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.pekko.projection.r2dbc + +import java.time.Instant +import java.util.UUID + +import org.apache.pekko +import pekko.actor.testkit.typed.scaladsl.LogCapturing +import pekko.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit +import pekko.persistence.Persistence +import pekko.persistence.query.TimestampOffset +import pekko.persistence.r2dbc.Dialect +import pekko.persistence.r2dbc.internal.R2dbcExecutor +import pekko.projection.ProjectionId +import pekko.projection.internal.ManagementState +import pekko.projection.r2dbc.R2dbcTimestampOffsetStoreSpec.TestTimestampSourceProvider +import pekko.projection.r2dbc.internal.OffsetPidSeqNr +import pekko.projection.r2dbc.internal.R2dbcOffsetStore +import pekko.projection.r2dbc.internal.R2dbcProjectionImpl +import pekko.projection.r2dbc.internal.mariadb.MariaDBR2dbcOffsetStore +import com.typesafe.config.ConfigFactory +import org.scalatest.wordspec.AnyWordSpecLike + +class MariaDBR2dbcOffsetStoreSpec + extends ScalaTestWithActorTestKit(TestConfig.config) + with AnyWordSpecLike + with LogCapturing { + + private val settings = R2dbcProjectionSettings(system) + + private def assumeMariaDB(): Unit = + assume(settings.dialect == Dialect.MariaDB, "requires a MariaDB database, run with the mariadb dialect") + + private def createR2dbcExecutor(settings: R2dbcProjectionSettings): R2dbcExecutor = + R2dbcProjectionImpl.createR2dbcExecutor( + R2dbcProjectionImpl.connectionFactory(settings, ConfigFactory.empty()), + settings, + ConfigFactory.empty()) + + private def createOffsetStore(projectionId: ProjectionId, settings: R2dbcProjectionSettings): R2dbcOffsetStore = { + val r2dbcExecutor = createR2dbcExecutor(settings) + val sourceProvider = + new TestTimestampSourceProvider(0, Persistence(system).numberOfSlices - 1, TestClock.nowMicros()) + R2dbcOffsetStore.fromConfig(projectionId, Some(sourceProvider), system, settings, r2dbcExecutor) + } + + private def genProjectionId() = ProjectionId(UUID.randomUUID().toString, "00") + + "The MariaDB offset store" must { + + "be created for the mariadb dialect, with upserts that don't use the MySQL row alias" in { + // no database calls, so this runs with every dialect + val offsetStore = createOffsetStore(genProjectionId(), settings.withDialect(Dialect.MariaDB)) match { + case mariaDB: MariaDBR2dbcOffsetStore => mariaDB + case other => fail(s"Expected MariaDBR2dbcOffsetStore, but was [${other.getClass.getName}]") + } + + (offsetStore.upsertOffsetSql should not).include("AS excluded") + offsetStore.upsertOffsetSql should include("current_offset = VALUES(current_offset)") + (offsetStore.updateManagementStateSql should not).include("AS excluded") + offsetStore.updateManagementStateSql should include("paused = VALUES(paused)") + } + + "update offsets and management state" in { + assumeMariaDB() + val offsetStore = createOffsetStore(genProjectionId(), settings) + + offsetStore.saveOffset(OffsetPidSeqNr(1L)).futureValue + offsetStore.saveOffset(OffsetPidSeqNr(2L)).futureValue + offsetStore.readOffset[Long]().futureValue shouldBe Some(2L) + + offsetStore.savePaused(paused = true).futureValue + offsetStore.readManagementState().futureValue shouldBe Some(ManagementState(paused = true)) + offsetStore.savePaused(paused = false).futureValue + offsetStore.readManagementState().futureValue shouldBe Some(ManagementState(paused = false)) + } + + "store timestamp offsets as the same instant, independent of the time zone of the server" in { + assumeMariaDB() + // The MariaDB container of the tests runs in America/New_York. Instants are bound as their UTC local time, + // which the server interprets in the session time zone. With a New York session, 01:30 on 2026-11-01 would be + // stored 5 hours later and still read back unchanged, so the stored instant is checked with UNIX_TIMESTAMP, + // which doesn't depend on the session time zone. 02:30 on 2027-03-14 doesn't exist in New York, the server + // would reject it. + val r2dbcExecutor = createR2dbcExecutor(settings) + val timestamps = Seq(Instant.parse("2026-11-01T01:30:00.123456Z"), Instant.parse("2027-03-14T02:30:00.654321Z")) + + timestamps.zipWithIndex.foreach { case (timestamp, i) => + val projectionId = genProjectionId() + val offsetStore = createOffsetStore(projectionId, settings) + val offset = TimestampOffset(timestamp, Map("p1" -> (i + 1L))) + offsetStore.saveOffset(OffsetPidSeqNr(offset, "p1", i + 1L)).futureValue + withClue(s"timestamp [$timestamp]: ") { + val storedEpochSeconds = r2dbcExecutor + .selectOne("select stored timestamp offset")( + _.createStatement( + s"SELECT UNIX_TIMESTAMP(timestamp_offset) AS epoch_seconds FROM ${settings.timestampOffsetTableWithSchema} " + + "WHERE projection_name = ?") + .bind(0, projectionId.name), + row => row.get("epoch_seconds", classOf[java.math.BigDecimal])) + .futureValue + storedEpochSeconds.map(BigDecimal(_)) shouldBe Some( + BigDecimal(timestamp.getEpochSecond) + BigDecimal(timestamp.getNano) / 1000000000) + + // a new offset store reads the offset from the database + createOffsetStore(projectionId, settings).readOffset[TimestampOffset]().futureValue shouldBe Some( + offset) + } + } + } + } +} diff --git a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/R2dbcProjectionSpec.scala b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/R2dbcProjectionSpec.scala index 5ee45856..ba0338f2 100644 --- a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/R2dbcProjectionSpec.scala +++ b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/R2dbcProjectionSpec.scala @@ -99,8 +99,8 @@ object R2dbcProjectionSpec { def createTableSql(dialect: Dialect): String = { val quotedTable = dialect match { // H2 would make an unquoted name upper case, so it must not be quoted in one place only - case Dialect.MySQL | Dialect.H2 => table - case _ => s""""$table"""" + case Dialect.MySQL | Dialect.MariaDB | Dialect.H2 => table + case _ => s""""$table"""" } s"""|CREATE table IF NOT EXISTS $quotedTable ( | id VARCHAR(255) NOT NULL, @@ -158,6 +158,13 @@ object R2dbcProjectionSpec { id = excluded.id, concatenated = excluded.concatenated """ + case Dialect.MariaDB => + sql""" + INSERT INTO $table (id, concatenated) VALUES (?, ?) + ON DUPLICATE KEY UPDATE + id = VALUES(id), + concatenated = VALUES(concatenated) + """ case Dialect.H2 => sql""" MERGE INTO $table (id, concatenated) KEY(id) VALUES (?, ?) diff --git a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/R2dbcTimestampOffsetProjectionSpec.scala b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/R2dbcTimestampOffsetProjectionSpec.scala index ca666910..3721f495 100644 --- a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/R2dbcTimestampOffsetProjectionSpec.scala +++ b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/R2dbcTimestampOffsetProjectionSpec.scala @@ -54,6 +54,7 @@ import pekko.projection.internal.SingleHandlerStrategy import pekko.projection.r2dbc.internal.R2dbcOffsetStore import pekko.projection.r2dbc.internal.R2dbcProjectionImpl import pekko.projection.r2dbc.internal.h2.H2R2dbcOffsetStore +import pekko.projection.r2dbc.internal.mariadb.MariaDBR2dbcOffsetStore import pekko.projection.r2dbc.internal.mysql.MySQLR2dbcOffsetStore import pekko.projection.r2dbc.scaladsl.R2dbcHandler import pekko.projection.r2dbc.scaladsl.R2dbcProjection @@ -1480,6 +1481,13 @@ class R2dbcTimestampOffsetProjectionSpec super.deleteOldTimestampOffsets() } } + case Dialect.MariaDB => + new MariaDBR2dbcOffsetStore(projectionId, Some(sourceProvider), system, spySettings, r2dbcExecutor) { + override def deleteOldTimestampOffsets(): Future[Long] = { + deleteCount.incrementAndGet() + super.deleteOldTimestampOffsets() + } + } case Dialect.H2 => new H2R2dbcOffsetStore(projectionId, Some(sourceProvider), system, spySettings, r2dbcExecutor) { override def deleteOldTimestampOffsets(): Future[Long] = { diff --git a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/TestConfig.scala b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/TestConfig.scala index 2760dc8b..68414afb 100644 --- a/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/TestConfig.scala +++ b/r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/TestConfig.scala @@ -59,6 +59,21 @@ object TestConfig { use-app-timestamp = on } """) + case "mariadb" => + ConfigFactory.parseString(""" + pekko.persistence.r2dbc { + connection-factory { + driver = "mariadb" + host = "localhost" + port = 3306 + user = "root" + password = "root" + database = "mysql" + } + db-timestamp-monotonic-increasing = on + use-app-timestamp = on + } + """) case "h2" => ConfigFactory.parseString(""" pekko.persistence.r2dbc { diff --git a/r2dbc/src/main/resources/reference.conf b/r2dbc/src/main/resources/reference.conf index 1aafbfc2..d506709d 100644 --- a/r2dbc/src/main/resources/reference.conf +++ b/r2dbc/src/main/resources/reference.conf @@ -5,7 +5,7 @@ //#projection-config pekko.projection.r2dbc { - # postgres, yugabyte, mysql or h2 + # postgres, yugabyte, mysql, mariadb or h2 dialect = ${pekko.persistence.r2dbc.dialect} offset-store { @@ -40,7 +40,7 @@ pekko.projection.r2dbc { # Batch size for inserts of timestamp offsets. Can be used to improve throughput for projections # with many events per persistence id and that is using `groupedWithinAsync`. # For Postgres (and Yugabyte) this will use multi-row inserts for this number of records. For MySQL - # this will use multiple single row inserts. + # and MariaDB this will use multiple single row inserts. # Use 0 to disable batching and always use single row inserts. offset-batch-size = 20 } diff --git a/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/R2dbcOffsetStore.scala b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/R2dbcOffsetStore.scala index f5fef713..63d2525c 100644 --- a/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/R2dbcOffsetStore.scala +++ b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/R2dbcOffsetStore.scala @@ -52,6 +52,7 @@ import pekko.projection.internal.OffsetSerialization.MultipleOffsets import pekko.projection.internal.OffsetSerialization.SingleOffset import pekko.projection.r2dbc.R2dbcProjectionSettings import pekko.projection.r2dbc.internal.h2.H2R2dbcOffsetStore +import pekko.projection.r2dbc.internal.mariadb.MariaDBR2dbcOffsetStore import pekko.projection.r2dbc.internal.mysql.MySQLR2dbcOffsetStore import io.r2dbc.spi.Connection import io.r2dbc.spi.Statement @@ -194,6 +195,8 @@ private[projection] object R2dbcOffsetStore { new R2dbcOffsetStore(projectionId, sourceProvider, system, settings, r2dbcExecutor, clock) case Dialect.MySQL => new MySQLR2dbcOffsetStore(projectionId, sourceProvider, system, settings, r2dbcExecutor, clock) + case Dialect.MariaDB => + new MariaDBR2dbcOffsetStore(projectionId, sourceProvider, system, settings, r2dbcExecutor, clock) case Dialect.H2 => new H2R2dbcOffsetStore(projectionId, sourceProvider, system, settings, r2dbcExecutor, clock) } diff --git a/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/mariadb/MariaDBR2dbcOffsetStore.scala b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/mariadb/MariaDBR2dbcOffsetStore.scala new file mode 100644 index 00000000..794b2b82 --- /dev/null +++ b/r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/mariadb/MariaDBR2dbcOffsetStore.scala @@ -0,0 +1,66 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.pekko.projection.r2dbc.internal.mariadb + +import java.time.Clock + +import org.apache.pekko +import pekko.actor.typed.ActorSystem +import pekko.annotation.InternalApi +import pekko.persistence.r2dbc.internal.R2dbcExecutor +import pekko.persistence.r2dbc.internal.Sql.DialectInterpolation +import pekko.projection.BySlicesSourceProvider +import pekko.projection.ProjectionId +import pekko.projection.r2dbc.R2dbcProjectionSettings +import pekko.projection.r2dbc.internal.mysql.MySQLR2dbcOffsetStore + +/** + * INTERNAL API + * + * MariaDB uses the MySQL offset store, except for the upserts: MariaDB doesn't support the `AS excluded` row alias of + * MySQL, so the inserted values are referenced with `VALUES()`. `Instant` values are converted by the connection + * factory of pekko-persistence-r2dbc, since the MariaDB driver doesn't support them. + */ +@InternalApi +private[projection] class MariaDBR2dbcOffsetStore( + projectionId: ProjectionId, + sourceProvider: Option[BySlicesSourceProvider], + system: ActorSystem[?], + settings: R2dbcProjectionSettings, + r2dbcExecutor: R2dbcExecutor, + clock: Clock = Clock.systemUTC()) + extends MySQLR2dbcOffsetStore(projectionId, sourceProvider, system, settings, r2dbcExecutor, clock) { + + override val upsertOffsetSql: String = sql""" + INSERT INTO $offsetTable + (projection_name, projection_key, current_offset, manifest, mergeable, last_updated) + VALUES (?,?,?,?,?,?) + ON DUPLICATE KEY UPDATE + current_offset = VALUES(current_offset), + manifest = VALUES(manifest), + mergeable = VALUES(mergeable), + last_updated = VALUES(last_updated)""" + + override val updateManagementStateSql: String = sql""" + INSERT INTO $managementTable + (projection_name, projection_key, paused, last_updated) + VALUES (?,?,?,?) + ON DUPLICATE KEY UPDATE + paused = VALUES(paused), + last_updated = VALUES(last_updated)""" +}