pjfanning commented on code in PR #534: URL: https://github.com/apache/pekko-persistence-r2dbc/pull/534#discussion_r4392038066
########## core/src/main/scala/org/apache/pekko/persistence/r2dbc/internal/mariadb/MariaDBConnections.scala: ########## @@ -0,0 +1,46 @@ +/* + * 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.persistence.r2dbc.internal.mariadb + +import java.util.Locale + +import org.apache.pekko +import pekko.annotation.InternalApi +import pekko.persistence.r2dbc.internal.ConnectionAdapter +import io.r2dbc.spi.ConnectionFactoryOptions +import io.r2dbc.spi.Option + +/** + * INTERNAL API + * + * Adapts the MariaDB R2DBC driver to the types used by the plugin. The driver doesn't support `java.time.Instant`, so + * it is bound and read as a UTC `LocalDateTime`. That requires the session time zone to be UTC, which is set with the + * [[TimezoneOption]] of the driver. + */ +@InternalApi +private[r2dbc] object MariaDBConnections extends ConnectionAdapter with ConnectionAdapter.InstantAsUtcLocalDateTime { + + val TimezoneOption: Option[String] = Option.valueOf("timezone") Review Comment: Question: does r2dbc-mariadb send this as `SET time_zone='UTC'` literally? MariaDB only resolves named zones when the `mysql.time_zone*` tables are loaded, which is not the case on many managed instances or with `MARIADB_INITDB_SKIP_TZINFO`, and there every connection would fail with `Unknown or incorrect time zone`. The official Docker image used in CI loads the tables, so CI wouldn't catch it. Using `"+00:00"` would avoid the dependency, unless the driver already normalizes `UTC` to an offset. ########## core/src/main/scala/org/apache/pekko/persistence/r2dbc/snapshot/mariadb/MariaDBSnapshotDao.scala: ########## @@ -0,0 +1,54 @@ +/* + * 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.persistence.r2dbc.snapshot.mariadb + +import scala.concurrent.ExecutionContext +import scala.concurrent.duration.FiniteDuration +import io.r2dbc.spi.ConnectionFactory +import org.apache.pekko +import pekko.actor.typed.ActorSystem +import pekko.annotation.InternalApi +import pekko.persistence.r2dbc.SnapshotSettings +import pekko.persistence.r2dbc.internal.Sql.DialectInterpolation +import pekko.persistence.r2dbc.snapshot.SnapshotDao + +/** + * INTERNAL API + */ +@InternalApi +private[r2dbc] class MariaDBSnapshotDao( + settings: SnapshotSettings, connectionFactory: ConnectionFactory, + closeCallsExceeding: Option[FiniteDuration] = None +)(implicit ec: ExecutionContext, system: ActorSystem[?]) + extends SnapshotDao(settings, connectionFactory, closeCallsExceeding) { Review Comment: Would it be better to extend `MySQLSnapshotDao` here and override only `upsertSql`? Today the MySQL overrides are small, but #515 (snapshots as starting points) adds a number of `MySQLSnapshotDao` overrides: tags bound as JSON strings, `UNIX_TIMESTAMP`-based bucket queries, `currentDbTimestamp`, and an upsert that includes the `db_timestamp`/`tags` columns. Extending the Postgres `SnapshotDao` means MariaDB would inherit the Postgres SQL (e.g. `extract(EPOCH from db_timestamp)::BIGINT`) and array tag binding once both PRs are in. Whichever of the two merges second will need to reconcile this. ########## core/src/main/scala/org/apache/pekko/persistence/r2dbc/snapshot/SnapshotDao.scala: ########## @@ -70,6 +71,8 @@ private[r2dbc] object SnapshotDao { new SnapshotDao(settings, connectionFactory, closeCallsExceeding) case Dialect.MySQL => new MySQLSnapshotDao(settings, connectionFactory, closeCallsExceeding) + case Dialect.MariaDB => Review Comment: The connection adapter is chosen from the driver option, while the DAO is chosen from the dialect, and nothing checks they agree. `dialect = mysql` with `driver = mariadb` gets the MySQL `AS excluded` upsert, which MariaDB rejects on the first snapshot save; `dialect = mariadb` with `driver = mysql` gets no UTC session or adapter. Would a fail-fast check in settings or `ConnectionFactoryProvider` for a mismatched dialect/driver be worth adding? ########## core/src/test/scala/org/apache/pekko/persistence/r2dbc/state/DurableStateStoreChangeHandlerSpec.scala: ########## @@ -53,9 +53,9 @@ object DurableStateStoreChangeHandlerSpec { val dialect: String = config.getString("pekko.persistence.r2dbc.dialect") - // MySQL uses ? as parameter markers; Postgres/Yugabyte use $1, $2, $3 + // MySQL and MariaDB use ? as parameter markers; Postgres/Yugabyte use $1, $2, $3 private val insertSql: String = - if (dialect == "mysql") "insert into changes_test (pid, rev, the_value) values (?, ?, ?)" + if (dialect == "mysql" || dialect == "mariadb") "insert into changes_test (pid, rev, the_value) values (?, ?, ?)" else sql"insert into changes_test (pid, rev, the_value) values (?, ?, ?)" Review Comment: This `dialect == "mysql" || dialect == "mariadb"` check is repeated in several specs (JavadslChangeHandler, PersistTagsSpec, EventsByPersistenceIdGapSpec, R2dbcExecutorSpec, DbTimestampDstSpec, here). A shared helper in `TestConfig` (or using `Sql.DialectInterpolation` with the parsed dialect) would mean the next dialect doesn't need to touch each spec. ########## docs/src/test/resources/application-mariadb.conf: ########## @@ -0,0 +1,33 @@ +# SPDX-License-Identifier: Apache-2.0 + +pekko.persistence.journal.plugin = "pekko.persistence.r2dbc.journal" +pekko.persistence.snapshot-store.plugin = "pekko.persistence.r2dbc.snapshot" +pekko.persistence.state.plugin = "pekko.persistence.r2dbc.state" + +// #connection-settings +pekko.persistence.r2dbc { + dialect = "mariadb" + connection-factory { + driver = "mariadb" + host = "localhost" + host = ${?DB_HOST} + port = 3306 + database = "mysql" + database = ${?DB_NAME} + user = "root" + user = ${?DB_USER} + password = "root" + password = ${?DB_PASSWORD} + + # ssl { Review Comment: These `ssl` settings map to the Postgres driver options (`PostgresqlConnectionFactoryProvider.SSL_MODE` / `SSL_ROOT_CERT`) in `ConnectionFactoryProvider`. I don't think r2dbc-mariadb reads them under those names/types (it seems to use `sslMode` with its own values and `serverSslCert`), so a user uncommenting this might end up without server certificate verification, or with a startup error. Could you check, and either map them for MariaDB or drop/adjust the commented block? ########## core/src/main/scala/org/apache/pekko/persistence/r2dbc/internal/mariadb/MariaDBConnections.scala: ########## @@ -0,0 +1,46 @@ +/* + * 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.persistence.r2dbc.internal.mariadb + +import java.util.Locale + +import org.apache.pekko +import pekko.annotation.InternalApi +import pekko.persistence.r2dbc.internal.ConnectionAdapter +import io.r2dbc.spi.ConnectionFactoryOptions +import io.r2dbc.spi.Option + +/** + * INTERNAL API + * + * Adapts the MariaDB R2DBC driver to the types used by the plugin. The driver doesn't support `java.time.Instant`, so + * it is bound and read as a UTC `LocalDateTime`. That requires the session time zone to be UTC, which is set with the + * [[TimezoneOption]] of the driver. + */ +@InternalApi +private[r2dbc] object MariaDBConnections extends ConnectionAdapter with ConnectionAdapter.InstantAsUtcLocalDateTime { + + val TimezoneOption: Option[String] = Option.valueOf("timezone") + + def isMariaDB(options: ConnectionFactoryOptions): Boolean = { + // the protocol is the driver when it is used through r2dbc-pool, e.g. r2dbc:pool:mariadb:// + def is(option: Option[?]): Boolean = + scala.Option(options.getValue(option)).exists(_.toString.toLowerCase(Locale.ROOT) == "mariadb") + is(ConnectionFactoryOptions.DRIVER) || is(ConnectionFactoryOptions.PROTOCOL) Review Comment: Detecting MariaDB from the protocol for `r2dbc:pool:mariadb://` means `ConnectionFactories.get` returns an r2dbc-pool `ConnectionPool`, which `ConnectionFactoryProvider` then wraps in its own `ConnectionPool`. The plugin's `max-size`, `acquire-timeout`, `max-life-time` etc. then don't govern the physical connections (the inner pool uses its own defaults), and the inner pool isn't disposed by `ConnectionFactoryProvider.close`. Is supporting this URL form intended? If not, it may be better not to encourage it (the spec currently asserts it). ########## core/src/main/scala/org/apache/pekko/persistence/r2dbc/ConnectionFactoryProvider.scala: ########## @@ -185,7 +186,15 @@ class ConnectionFactoryProvider(system: ActorSystem[?]) extends Extension { builder.option(Option.valueOf("connectionTimeZone"), "SERVER") } - ConnectionFactories.get(customizer(builder, config).build()) + if (MariaDBConnections.isMariaDB(builder.build())) { + // Instants are bound and read as UTC LocalDateTime, see MariaDBConnections + builder.option(MariaDBConnections.TimezoneOption, "UTC") + } + + val options = customizer(builder, config).build() + val connectionFactory = ConnectionFactories.get(options) Review Comment: The UTC `timezone` option is set before the user's `connection-factory-options-customizer` runs, but `MariaDBConnections.wrap` is then applied unconditionally. If a customizer sets `timezone` to `auto`/`SYSTEM` (or rebuilds the options), Instants are still bound as UTC `LocalDateTime` while the session uses the server zone, so on a non-UTC server every `db_timestamp` is silently shifted. Could we either set the option after the customizer, or check `options.getValue(MariaDBConnections.TimezoneOption)` is UTC after `customizer(...)` and fail fast otherwise? ########## core/src/main/scala/org/apache/pekko/persistence/r2dbc/internal/mariadb/MariaDBConnections.scala: ########## @@ -0,0 +1,46 @@ +/* + * 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.persistence.r2dbc.internal.mariadb + +import java.util.Locale + +import org.apache.pekko +import pekko.annotation.InternalApi +import pekko.persistence.r2dbc.internal.ConnectionAdapter +import io.r2dbc.spi.ConnectionFactoryOptions +import io.r2dbc.spi.Option + +/** + * INTERNAL API + * + * Adapts the MariaDB R2DBC driver to the types used by the plugin. The driver doesn't support `java.time.Instant`, so + * it is bound and read as a UTC `LocalDateTime`. That requires the session time zone to be UTC, which is set with the + * [[TimezoneOption]] of the driver. + */ +@InternalApi +private[r2dbc] object MariaDBConnections extends ConnectionAdapter with ConnectionAdapter.InstantAsUtcLocalDateTime { Review Comment: Nit: `ConnectionAdapter.InstantAsUtcLocalDateTime` already extends `ConnectionAdapter`, so this could be `extends ConnectionAdapter.InstantAsUtcLocalDateTime`. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
