This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-persistence-r2dbc.git
The following commit(s) were added to refs/heads/main by this push:
new 48a6651 Close connection if execution takes too long (#457)
48a6651 is described below
commit 48a6651dce1605d11af0bdc368fd1e3f7e236d16
Author: PJ Fanning <[email protected]>
AuthorDate: Mon Jul 27 10:43:26 2026 +0100
Close connection if execution takes too long (#457)
* Close connection if execution takes too long
* Update R2dbcExecutorSpec.scala
* Update R2dbcExecutor.scala
* Update R2dbcExecutorSpec.scala
* Update R2dbcExecutor.scala
* try to fix plugin test issue
* scalafmt
* remove blank line
---
core/src/main/resources/reference.conf | 18 +++
.../r2dbc/ConnectionFactoryProvider.scala | 4 +
.../pekko/persistence/r2dbc/R2dbcSettings.scala | 14 +++
.../persistence/r2dbc/internal/R2dbcExecutor.scala | 106 +++++++++++++----
.../persistence/r2dbc/journal/JournalDao.scala | 16 ++-
.../r2dbc/journal/mysql/MySQLJournalDao.scala | 6 +-
.../r2dbc/query/scaladsl/QueryDao.scala | 14 ++-
.../r2dbc/query/scaladsl/mysql/MySQLQueryDao.scala | 6 +-
.../r2dbc/session/scaladsl/R2dbcSession.scala | 8 +-
.../persistence/r2dbc/snapshot/SnapshotDao.scala | 16 ++-
.../r2dbc/snapshot/mysql/MySQLSnapshotDao.scala | 7 +-
.../r2dbc/state/scaladsl/DurableStateDao.scala | 15 ++-
.../scaladsl/mysql/MySQLDurableStateDao.scala | 6 +-
.../pekko/persistence/r2dbc/TestDbLifecycle.scala | 9 +-
.../r2dbc/internal/R2dbcExecutorSpec.scala | 131 +++++++++++++++++++++
.../r2dbc/migration/MigrationTool.scala | 7 +-
.../r2dbc/migration/MigrationToolDao.scala | 6 +-
17 files changed, 338 insertions(+), 51 deletions(-)
diff --git a/core/src/main/resources/reference.conf
b/core/src/main/resources/reference.conf
index b05b9aa..35c9bc1 100644
--- a/core/src/main/resources/reference.conf
+++ b/core/src/main/resources/reference.conf
@@ -283,6 +283,24 @@ pekko.persistence.r2dbc {
# A fast query for Postgres is "SELECT 1"
validation-query = ""
+ # Abort any statement that takes more than the specified amount of time.
+ # This timeout is handled by the database server.
+ # This timeout should be less than `close-calls-exceeding`.
+ statement-timeout = off
+
+ # Maximum SQL statement execution duration. The current connection is
closed if exceeded,
+ # and will not be reused by the pool.
+ # This timeout is handled on the client side and should be used in case
the database server
+ # is unresponsive or the connection is broken but not closed.
+ # It can be used in combination with `statement-timeout`, which should be
less than this
+ # timeout.
+ # The timeout is needed to handle some failure scenarios, when the
database server is
+ # terminated in a non graceful way and there is a load balancer in front.
The client
+ # connection and current execution in progress would not be completed
without this timeout,
+ # resulting in a "connection leak".
+ # Set to "off" to disable this timeout.
+ close-calls-exceeding = 20 seconds
+
# FQCN of a ConnectionFactoryOptionsCustomizer. If non-empty, it must be
the fully
# qualified class name of a class implementing the trait
ConnectionFactoryOptionsCustomizer.
# The class must have a constructor with a single parameter of type
ActorSystem[_].
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/r2dbc/ConnectionFactoryProvider.scala
b/core/src/main/scala/org/apache/pekko/persistence/r2dbc/ConnectionFactoryProvider.scala
index 2d9a1d8..a65c410 100644
---
a/core/src/main/scala/org/apache/pekko/persistence/r2dbc/ConnectionFactoryProvider.scala
+++
b/core/src/main/scala/org/apache/pekko/persistence/r2dbc/ConnectionFactoryProvider.scala
@@ -147,6 +147,10 @@ class ConnectionFactoryProvider(system: ActorSystem[?])
extends Extension {
PostgresqlConnectionFactoryProvider.PREPARED_STATEMENT_CACHE_QUERIES,
Integer.valueOf(settings.statementCacheSize))
+ settings.statementTimeout.foreach { timeout =>
+ builder.option(PostgresqlConnectionFactoryProvider.STATEMENT_TIMEOUT,
JDuration.ofMillis(timeout.toMillis))
+ }
+
if (settings.sslEnabled) {
builder.option(ConnectionFactoryOptions.SSL, java.lang.Boolean.TRUE)
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/r2dbc/R2dbcSettings.scala
b/core/src/main/scala/org/apache/pekko/persistence/r2dbc/R2dbcSettings.scala
index 76c5630..81d9357 100644
--- a/core/src/main/scala/org/apache/pekko/persistence/r2dbc/R2dbcSettings.scala
+++ b/core/src/main/scala/org/apache/pekko/persistence/r2dbc/R2dbcSettings.scala
@@ -331,6 +331,14 @@ final class ConnectionFactorySettings(config: Config) {
val statementCacheSize: Int = config.getInt("statement-cache-size")
+ val statementTimeout: Option[FiniteDuration] =
+ config.getString("statement-timeout").toLowerCase(Locale.ROOT) match {
+ case "off" => None
+ case _ => Some(config.getDuration("statement-timeout").toScala)
+ }
+
+ val closeCallsExceeding: Option[FiniteDuration] =
ConnectionFactorySettings.closeCallsExceeding(config)
+
val connectionFactoryOptionsCustomizer: Option[String] =
Option(config.getString("connection-factory-options-customizer")).filter(_.trim.nonEmpty)
}
@@ -342,6 +350,12 @@ final class ConnectionFactorySettings(config: Config) {
object ConnectionFactorySettings {
def apply(config: Config): ConnectionFactorySettings =
new ConnectionFactorySettings(config)
+
+ def closeCallsExceeding(config: Config): Option[FiniteDuration] =
+ config.getString("close-calls-exceeding").toLowerCase(Locale.ROOT) match {
+ case "off" => None
+ case _ => Some(config.getDuration("close-calls-exceeding").toScala)
+ }
}
/**
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/r2dbc/internal/R2dbcExecutor.scala
b/core/src/main/scala/org/apache/pekko/persistence/r2dbc/internal/R2dbcExecutor.scala
index 4d62500..a2808a0 100644
---
a/core/src/main/scala/org/apache/pekko/persistence/r2dbc/internal/R2dbcExecutor.scala
+++
b/core/src/main/scala/org/apache/pekko/persistence/r2dbc/internal/R2dbcExecutor.scala
@@ -19,7 +19,11 @@ import scala.collection.immutable
import scala.collection.mutable
import scala.concurrent.ExecutionContext
import scala.concurrent.Future
+import scala.concurrent.duration.DurationInt
import scala.concurrent.duration.FiniteDuration
+import scala.util.Failure
+import scala.util.Success
+import scala.util.control.Exception.Catcher
import scala.util.control.NonFatal
import org.apache.pekko
@@ -31,6 +35,7 @@ import io.r2dbc.spi.ConnectionFactory
import io.r2dbc.spi.Result
import io.r2dbc.spi.Row
import io.r2dbc.spi.Statement
+import io.r2dbc.spi.Wrapped
import org.reactivestreams.Publisher
import org.slf4j.Logger
import reactor.core.publisher.Flux
@@ -108,15 +113,25 @@ import reactor.core.publisher.Mono
* INTERNAL API:
*/
@InternalStableApi
-class R2dbcExecutor(val connectionFactory: ConnectionFactory, log: Logger,
logDbCallsExceeding: FiniteDuration)(
- implicit
- ec: ExecutionContext,
- system: ActorSystem[?]) {
+class R2dbcExecutor(
+ val connectionFactory: ConnectionFactory,
+ log: Logger,
+ logDbCallsExceeding: FiniteDuration,
+ closeCallsExceeding: Option[FiniteDuration])(implicit ec:
ExecutionContext, system: ActorSystem[?]) {
+
import R2dbcExecutor._
private val logDbCallsExceedingMicros = logDbCallsExceeding.toMicros
private val logDbCallsExceedingEnabled = logDbCallsExceedingMicros >= 0
+ // for backwards compatibility, closeCallsExceeding should be defined
+ @deprecated("Use constructor with closeCallsExceeding", "2.0.0")
+ def this(connectionFactory: ConnectionFactory, log: Logger,
logDbCallsExceeding: FiniteDuration)(
+ implicit
+ ec: ExecutionContext,
+ system: ActorSystem[?]) =
+ this(connectionFactory, log, logDbCallsExceeding, closeCallsExceeding =
Some(20.seconds))
+
private def nanoTime(): Long =
if (logDbCallsExceedingEnabled) System.nanoTime() else 0L
@@ -271,6 +286,10 @@ class R2dbcExecutor(val connectionFactory:
ConnectionFactory, log: Logger, logDb
getConnection(logPrefix).flatMap { connection =>
val startTime = nanoTime()
connection.beginTransaction().asFutureDone().flatMap { _ =>
+ val timeoutTask = closeCallsExceeding.map { timeout =>
+ system.scheduler.scheduleOnce(timeout, () =>
closeAfterTimeout(connection))
+ }
+
val result =
try {
fun(connection)
@@ -284,16 +303,20 @@ class R2dbcExecutor(val connectionFactory:
ConnectionFactory, log: Logger, logDb
if (log.isDebugEnabled())
log.debug("{} - DB call failed: {}", logPrefix: Any, exc.toString:
Any)
// ok to rollback async like this, or should it be before completing
the returned Future?
- rollbackAndClose(connection)
+ val done = rollbackAndClose(connection)
+ timeoutTask.foreach { task => done.onComplete(_ => task.cancel()) }
}
result.flatMap { r =>
- commitAndClose(connection).map { _ =>
- val durationMicros = durationInMicros(startTime)
- if (durationMicros >= logDbCallsExceedingMicros)
- log.info("{} - DB call completed in [{}] µs", logPrefix,
durationMicros)
- r
- }
+ val done = commitAndClose(connection)
+ .map { _ =>
+ val durationMicros = durationInMicros(startTime)
+ if (durationMicros >= logDbCallsExceedingMicros)
+ log.info("{} - DB call completed in [{}] µs", logPrefix,
durationMicros)
+ r
+ }
+ timeoutTask.foreach { task => done.onComplete(_ => task.cancel()) }
+ done
}
}
@@ -306,6 +329,10 @@ class R2dbcExecutor(val connectionFactory:
ConnectionFactory, log: Logger, logDb
def withAutoCommitConnection[A](logPrefix: String)(fun: Connection =>
Future[A]): Future[A] = {
getConnection(logPrefix).flatMap { connection =>
val startTime = nanoTime()
+ val timeoutTask = closeCallsExceeding.map { timeout =>
+ system.scheduler.scheduleOnce(timeout, () =>
closeAfterTimeout(connection))
+ }
+
connection.setAutoCommit(true).asFutureDone().flatMap { _ =>
val result =
try {
@@ -319,27 +346,64 @@ class R2dbcExecutor(val connectionFactory:
ConnectionFactory, log: Logger, logDb
result.failed.foreach { exc =>
log.debug("{} - DB call failed: {}", logPrefix: Any, exc: Any)
// auto-commit so nothing to rollback
- connection.close().asFutureDone()
+ val done = connection.close().asFutureDone()
+ timeoutTask.foreach { task => done.onComplete(_ => task.cancel()) }
}
result.flatMap { r =>
- connection.close().asFutureDone().map { _ =>
- val durationMicros = durationInMicros(startTime)
- if (durationMicros >= logDbCallsExceedingMicros)
- log.info("{} - DB call completed [{}] in [{}] µs", logPrefix,
Option(r).map(_.toString).orNull,
- durationMicros: java.lang.Long)
- r
- }
+ val done = connection
+ .close()
+ .asFutureDone()
+ .map { _ =>
+ val durationMicros = durationInMicros(startTime)
+ if (durationMicros >= logDbCallsExceedingMicros)
+ log.info("{} - DB call completed [{}] in [{}] µs", logPrefix,
Option(r).map(_.toString).orNull,
+ durationMicros: java.lang.Long)
+ r
+ }
+ timeoutTask.foreach { task => done.onComplete(_ => task.cancel()) }
+ done
}
}
}
}
private def commitAndClose(connection: Connection): Future[Done] = {
- connection.commitTransaction().asFutureDone().andThen { case _ =>
connection.close().asFutureDone() }
+ connection.commitTransaction().asFutureDone().andThen { case _ =>
+ try connection.close().asFutureDone()
+ catch ignoreConnectionClosedException
+ }
}
private def rollbackAndClose(connection: Connection): Future[Done] = {
- connection.rollbackTransaction().asFutureDone().andThen { case _ =>
connection.close().asFutureDone() }
+ try connection
+ .rollbackTransaction()
+ .asFutureDone()
+ .andThen { case _ =>
+ try connection.close().asFutureDone()
+ catch ignoreConnectionClosedException
+ }
+ catch ignoreConnectionClosedException
+ }
+
+ private def closeAfterTimeout(connection: Connection): Unit = {
+ log.warn("Timeout after [{}], will try to close connection",
closeCallsExceeding.map(_.toCoarsest).getOrElse(""))
+
+ val done = connection match {
+ case wrapped: Wrapped[Connection] @unchecked =>
wrapped.unwrap().close.asFutureDone()
+ case _ =>
connection.close.asFutureDone()
+ }
+
+ done.onComplete {
+ case Success(_) =>
+ log.debug("Connection closed after timeout")
+ case Failure(exc) =>
+ log.debug("Tried to close connection after timeout, but {}",
exc.toString)
+ }
+ }
+
+ private def ignoreConnectionClosedException: Catcher[Future[Done]] = { case
_: IllegalStateException =>
+ // throw by PooledConnection assertNotClosed, if connection closed after
timeout
+ Future.successful(Done)
}
}
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/r2dbc/journal/JournalDao.scala
b/core/src/main/scala/org/apache/pekko/persistence/r2dbc/journal/JournalDao.scala
index afc5867..a66fb52 100644
---
a/core/src/main/scala/org/apache/pekko/persistence/r2dbc/journal/JournalDao.scala
+++
b/core/src/main/scala/org/apache/pekko/persistence/r2dbc/journal/JournalDao.scala
@@ -16,12 +16,14 @@ package org.apache.pekko.persistence.r2dbc.journal
import java.time.Instant
import scala.concurrent.{ ExecutionContext, Future }
+import scala.concurrent.duration.FiniteDuration
import org.apache.pekko
import pekko.actor.typed.ActorSystem
import pekko.annotation.InternalApi
import pekko.persistence.Persistence
import pekko.persistence.r2dbc.ConnectionFactoryProvider
+import pekko.persistence.r2dbc.ConnectionFactorySettings
import pekko.persistence.r2dbc.Dialect
import pekko.persistence.r2dbc.JournalSettings
import pekko.persistence.r2dbc.internal.BySliceQuery
@@ -84,11 +86,15 @@ private[r2dbc] object JournalDao {
)(implicit system: ActorSystem[?], ec: ExecutionContext): JournalDao = {
val connectionFactory =
ConnectionFactoryProvider(system).connectionFactoryFor(settings.useConnectionFactory,
config)
+ val closeCallsExceeding = {
+ val fullConfig = config.withFallback(system.settings.config)
+
ConnectionFactorySettings.closeCallsExceeding(fullConfig.getConfig(settings.useConnectionFactory))
+ }
settings.dialect match {
case Dialect.Postgres | Dialect.Yugabyte =>
- new JournalDao(settings, connectionFactory)
+ new JournalDao(settings, connectionFactory, closeCallsExceeding)
case Dialect.MySQL =>
- new MySQLJournalDao(settings, connectionFactory)
+ new MySQLJournalDao(settings, connectionFactory, closeCallsExceeding)
}
}
}
@@ -99,7 +105,8 @@ private[r2dbc] object JournalDao {
* Class for doing db interaction outside of an actor to avoid mistakes in
future callbacks
*/
@InternalApi
-private[r2dbc] class JournalDao(val settings: JournalSettings,
connectionFactory: ConnectionFactory)(
+private[r2dbc] class JournalDao(val settings: JournalSettings,
connectionFactory: ConnectionFactory,
+ closeCallsExceeding: Option[FiniteDuration] = None)(
implicit val ec: ExecutionContext, system: ActorSystem[?]) extends
EventsByPersistenceIdDao
with HighestSequenceNrDao {
import JournalDao.SerializedJournalRow
@@ -111,7 +118,8 @@ private[r2dbc] class JournalDao(val settings:
JournalSettings, connectionFactory
private val persistenceExt = Persistence(system)
- protected val r2dbcExecutor = new R2dbcExecutor(connectionFactory, log,
settings.logDbCallsExceeding)(ec, system)
+ protected val r2dbcExecutor =
+ new R2dbcExecutor(connectionFactory, log, settings.logDbCallsExceeding,
closeCallsExceeding)(ec, system)
protected val journalTable: String = settings.journalTableWithSchema
protected implicit val journalPayloadCodec: PayloadCodec =
settings.journalPayloadCodec
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/r2dbc/journal/mysql/MySQLJournalDao.scala
b/core/src/main/scala/org/apache/pekko/persistence/r2dbc/journal/mysql/MySQLJournalDao.scala
index 69b1476..be5226c 100644
---
a/core/src/main/scala/org/apache/pekko/persistence/r2dbc/journal/mysql/MySQLJournalDao.scala
+++
b/core/src/main/scala/org/apache/pekko/persistence/r2dbc/journal/mysql/MySQLJournalDao.scala
@@ -20,6 +20,7 @@
package org.apache.pekko.persistence.r2dbc.journal.mysql
import scala.concurrent.ExecutionContext
+import scala.concurrent.duration.FiniteDuration
import org.apache.pekko
import pekko.actor.typed.ActorSystem
import pekko.annotation.InternalApi
@@ -118,9 +119,10 @@ private[r2dbc] object MySQLJournalDao {
@InternalApi
private[r2dbc] class MySQLJournalDao(
journalSettings: JournalSettings,
- connectionFactory: ConnectionFactory)(
+ connectionFactory: ConnectionFactory,
+ closeCallsExceeding: Option[FiniteDuration] = None)(
implicit ec: ExecutionContext, system: ActorSystem[?]
-) extends JournalDao(journalSettings, connectionFactory) {
+) extends JournalDao(journalSettings, connectionFactory, closeCallsExceeding) {
MySQLJournalDao.settingRequirements(journalSettings)
override lazy val timestampSql: String = "NOW(6)"
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/r2dbc/query/scaladsl/QueryDao.scala
b/core/src/main/scala/org/apache/pekko/persistence/r2dbc/query/scaladsl/QueryDao.scala
index 69c8aa6..9052db1 100644
---
a/core/src/main/scala/org/apache/pekko/persistence/r2dbc/query/scaladsl/QueryDao.scala
+++
b/core/src/main/scala/org/apache/pekko/persistence/r2dbc/query/scaladsl/QueryDao.scala
@@ -24,6 +24,7 @@ import pekko.NotUsed
import pekko.actor.typed.ActorSystem
import pekko.annotation.InternalApi
import pekko.persistence.r2dbc.ConnectionFactoryProvider
+import pekko.persistence.r2dbc.ConnectionFactorySettings
import pekko.persistence.r2dbc.Dialect
import pekko.persistence.r2dbc.QuerySettings
import pekko.persistence.r2dbc.internal.BySliceQuery
@@ -61,11 +62,15 @@ object QueryDao {
)(implicit system: ActorSystem[?], ec: ExecutionContext): QueryDao = {
val connectionFactory =
ConnectionFactoryProvider(system).connectionFactoryFor(settings.useConnectionFactory,
config)
+ val closeCallsExceeding = {
+ val fullConfig = config.withFallback(system.settings.config)
+
ConnectionFactorySettings.closeCallsExceeding(fullConfig.getConfig(settings.useConnectionFactory))
+ }
settings.dialect match {
case Dialect.Postgres | Dialect.Yugabyte =>
- new QueryDao(settings, connectionFactory)
+ new QueryDao(settings, connectionFactory, closeCallsExceeding)
case Dialect.MySQL =>
- new MySQLQueryDao(settings, connectionFactory)
+ new MySQLQueryDao(settings, connectionFactory, closeCallsExceeding)
}
}
}
@@ -74,7 +79,8 @@ object QueryDao {
* INTERNAL API
*/
@InternalApi
-private[r2dbc] class QueryDao(val settings: QuerySettings, connectionFactory:
ConnectionFactory)(
+private[r2dbc] class QueryDao(val settings: QuerySettings, connectionFactory:
ConnectionFactory,
+ closeCallsExceeding: Option[FiniteDuration] = None)(
implicit val ec: ExecutionContext, system: ActorSystem[?]) extends
BySliceQuery.Dao[SerializedJournalRow]
with EventsByPersistenceIdDao with HighestSequenceNrDao {
import JournalDao.readMetadata
@@ -163,7 +169,7 @@ private[r2dbc] class QueryDao(val settings: QuerySettings,
connectionFactory: Co
sql"SELECT DISTINCT(persistence_id) from $journalTable WHERE
persistence_id LIKE ? AND persistence_id > ? ORDER BY persistence_id LIMIT ?"
protected val r2dbcExecutor =
- new R2dbcExecutor(connectionFactory, log,
settings.logDbCallsExceeding)(ec, system)
+ new R2dbcExecutor(connectionFactory, log, settings.logDbCallsExceeding,
closeCallsExceeding)(ec, system)
def currentDbTimestamp(): Future[Instant] = {
r2dbcExecutor
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/r2dbc/query/scaladsl/mysql/MySQLQueryDao.scala
b/core/src/main/scala/org/apache/pekko/persistence/r2dbc/query/scaladsl/mysql/MySQLQueryDao.scala
index 5a1bc24..50c25a6 100644
---
a/core/src/main/scala/org/apache/pekko/persistence/r2dbc/query/scaladsl/mysql/MySQLQueryDao.scala
+++
b/core/src/main/scala/org/apache/pekko/persistence/r2dbc/query/scaladsl/mysql/MySQLQueryDao.scala
@@ -41,8 +41,10 @@ import io.r2dbc.spi.Row
@InternalApi
private[r2dbc] class MySQLQueryDao(
querySettings: QuerySettings,
- connectionFactory: ConnectionFactory
-)(implicit ec: ExecutionContext, system: ActorSystem[?]) extends
QueryDao(querySettings, connectionFactory) {
+ connectionFactory: ConnectionFactory,
+ closeCallsExceeding: Option[FiniteDuration] = None
+)(implicit ec: ExecutionContext, system: ActorSystem[?])
+ extends QueryDao(querySettings, connectionFactory, closeCallsExceeding) {
override lazy val statementTimestampSql: String = "NOW(6)"
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/r2dbc/session/scaladsl/R2dbcSession.scala
b/core/src/main/scala/org/apache/pekko/persistence/r2dbc/session/scaladsl/R2dbcSession.scala
index 9c6d5fa..0974f61 100644
---
a/core/src/main/scala/org/apache/pekko/persistence/r2dbc/session/scaladsl/R2dbcSession.scala
+++
b/core/src/main/scala/org/apache/pekko/persistence/r2dbc/session/scaladsl/R2dbcSession.scala
@@ -22,6 +22,7 @@ import org.apache.pekko
import pekko.actor.typed.ActorSystem
import pekko.annotation.ApiMayChange
import pekko.persistence.r2dbc.ConnectionFactoryProvider
+import pekko.persistence.r2dbc.ConnectionFactorySettings
import pekko.persistence.r2dbc.internal.R2dbcExecutor
import io.r2dbc.spi.Connection
import io.r2dbc.spi.Row
@@ -45,7 +46,12 @@ object R2dbcSession {
def withSession[A](system: ActorSystem[?], connectionFactoryConfigPath:
String)(
fun: R2dbcSession => Future[A]): Future[A] = {
val connectionFactory =
ConnectionFactoryProvider(system).connectionFactoryFor(connectionFactoryConfigPath)
- val r2dbcExecutor = new R2dbcExecutor(connectionFactory, log,
logDbCallsDisabled)(system.executionContext, system)
+ val closeCallsExceeding =
+
ConnectionFactorySettings.closeCallsExceeding(system.settings.config.getConfig(connectionFactoryConfigPath))
+ val r2dbcExecutor =
+ new R2dbcExecutor(connectionFactory, log, logDbCallsDisabled,
closeCallsExceeding)(
+ system.executionContext,
+ system)
r2dbcExecutor.withConnection("R2dbcSession") { connection =>
val session = new R2dbcSession(connection)(system.executionContext,
system)
fun(session)
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/r2dbc/snapshot/SnapshotDao.scala
b/core/src/main/scala/org/apache/pekko/persistence/r2dbc/snapshot/SnapshotDao.scala
index e78b321..688e4eb 100644
---
a/core/src/main/scala/org/apache/pekko/persistence/r2dbc/snapshot/SnapshotDao.scala
+++
b/core/src/main/scala/org/apache/pekko/persistence/r2dbc/snapshot/SnapshotDao.scala
@@ -14,6 +14,7 @@
package org.apache.pekko.persistence.r2dbc.snapshot
import scala.concurrent.{ ExecutionContext, Future }
+import scala.concurrent.duration.FiniteDuration
import org.apache.pekko
import pekko.actor.typed.ActorSystem
@@ -21,6 +22,7 @@ import pekko.annotation.InternalApi
import pekko.persistence.Persistence
import pekko.persistence.SnapshotSelectionCriteria
import pekko.persistence.r2dbc.ConnectionFactoryProvider
+import pekko.persistence.r2dbc.ConnectionFactorySettings
import pekko.persistence.r2dbc.Dialect
import pekko.persistence.r2dbc.SnapshotSettings
import pekko.persistence.r2dbc.internal.PayloadCodec
@@ -59,11 +61,15 @@ private[r2dbc] object SnapshotDao {
)(implicit system: ActorSystem[?], ec: ExecutionContext): SnapshotDao = {
val connectionFactory =
ConnectionFactoryProvider(system).connectionFactoryFor(settings.useConnectionFactory,
config)
+ val closeCallsExceeding = {
+ val fullConfig = config.withFallback(system.settings.config)
+
ConnectionFactorySettings.closeCallsExceeding(fullConfig.getConfig(settings.useConnectionFactory))
+ }
settings.dialect match {
case Dialect.Postgres | Dialect.Yugabyte =>
- new SnapshotDao(settings, connectionFactory)
+ new SnapshotDao(settings, connectionFactory, closeCallsExceeding)
case Dialect.MySQL =>
- new MySQLSnapshotDao(settings, connectionFactory)
+ new MySQLSnapshotDao(settings, connectionFactory, closeCallsExceeding)
}
}
}
@@ -74,7 +80,8 @@ private[r2dbc] object SnapshotDao {
* Class for doing db interaction outside of an actor to avoid mistakes in
future callbacks
*/
@InternalApi
-private[r2dbc] class SnapshotDao(settings: SnapshotSettings,
connectionFactory: ConnectionFactory)(
+private[r2dbc] class SnapshotDao(settings: SnapshotSettings,
connectionFactory: ConnectionFactory,
+ closeCallsExceeding: Option[FiniteDuration] = None)(
implicit
ec: ExecutionContext,
system: ActorSystem[?]) {
@@ -84,7 +91,8 @@ private[r2dbc] class SnapshotDao(settings: SnapshotSettings,
connectionFactory:
protected val snapshotTable: String = settings.snapshotsTableWithSchema
private val persistenceExt = Persistence(system)
- private val r2dbcExecutor = new R2dbcExecutor(connectionFactory, log,
settings.logDbCallsExceeding)(ec, system)
+ private val r2dbcExecutor =
+ new R2dbcExecutor(connectionFactory, log, settings.logDbCallsExceeding,
closeCallsExceeding)(ec, system)
protected implicit val snapshotPayloadCodec: PayloadCodec =
settings.snapshotPayloadCodec
private def collectSerializedSnapshot(row: Row): SerializedSnapshotRow =
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/r2dbc/snapshot/mysql/MySQLSnapshotDao.scala
b/core/src/main/scala/org/apache/pekko/persistence/r2dbc/snapshot/mysql/MySQLSnapshotDao.scala
index ce92df6..75c9743 100644
---
a/core/src/main/scala/org/apache/pekko/persistence/r2dbc/snapshot/mysql/MySQLSnapshotDao.scala
+++
b/core/src/main/scala/org/apache/pekko/persistence/r2dbc/snapshot/mysql/MySQLSnapshotDao.scala
@@ -20,6 +20,7 @@
package org.apache.pekko.persistence.r2dbc.snapshot.mysql
import scala.concurrent.ExecutionContext
+import scala.concurrent.duration.FiniteDuration
import io.r2dbc.spi.ConnectionFactory
import org.apache.pekko
import pekko.actor.typed.ActorSystem
@@ -33,8 +34,10 @@ import pekko.persistence.r2dbc.snapshot.SnapshotDao
*/
@InternalApi
private[r2dbc] class MySQLSnapshotDao(
- settings: SnapshotSettings, connectionFactory: ConnectionFactory
-)(implicit ec: ExecutionContext, system: ActorSystem[?]) extends
SnapshotDao(settings, connectionFactory) {
+ settings: SnapshotSettings, connectionFactory: ConnectionFactory,
+ closeCallsExceeding: Option[FiniteDuration] = None
+)(implicit ec: ExecutionContext, system: ActorSystem[?])
+ extends SnapshotDao(settings, connectionFactory, closeCallsExceeding) {
override val upsertSql = sql"""
INSERT INTO $snapshotTable
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/r2dbc/state/scaladsl/DurableStateDao.scala
b/core/src/main/scala/org/apache/pekko/persistence/r2dbc/state/scaladsl/DurableStateDao.scala
index 93d63e2..5dee155 100644
---
a/core/src/main/scala/org/apache/pekko/persistence/r2dbc/state/scaladsl/DurableStateDao.scala
+++
b/core/src/main/scala/org/apache/pekko/persistence/r2dbc/state/scaladsl/DurableStateDao.scala
@@ -33,6 +33,7 @@ import pekko.persistence.query.DurableStateChange
import pekko.persistence.query.NoOffset
import pekko.persistence.query.UpdatedDurableState
import pekko.persistence.r2dbc.ConnectionFactoryProvider
+import pekko.persistence.r2dbc.ConnectionFactorySettings
import pekko.persistence.r2dbc.Dialect
import pekko.persistence.r2dbc.StateSettings
import pekko.persistence.r2dbc.internal.AdditionalColumnFactory
@@ -94,11 +95,15 @@ import org.slf4j.LoggerFactory
)(implicit system: ActorSystem[?], ec: ExecutionContext): DurableStateDao = {
val connectionFactory =
ConnectionFactoryProvider(system).connectionFactoryFor(settings.useConnectionFactory,
config)
+ val closeCallsExceeding = {
+ val fullConfig = config.withFallback(system.settings.config)
+
ConnectionFactorySettings.closeCallsExceeding(fullConfig.getConfig(settings.useConnectionFactory))
+ }
settings.dialect match {
case Dialect.Postgres | Dialect.Yugabyte =>
- new DurableStateDao(settings, connectionFactory)
+ new DurableStateDao(settings, connectionFactory, closeCallsExceeding)
case Dialect.MySQL =>
- new MySQLDurableStateDao(settings, connectionFactory)
+ new MySQLDurableStateDao(settings, connectionFactory,
closeCallsExceeding)
}
}
}
@@ -109,7 +114,8 @@ import org.slf4j.LoggerFactory
* Class for encapsulating db interaction.
*/
@InternalApi
-private[r2dbc] class DurableStateDao(settings: StateSettings,
connectionFactory: ConnectionFactory)(
+private[r2dbc] class DurableStateDao(settings: StateSettings,
connectionFactory: ConnectionFactory,
+ closeCallsExceeding: Option[FiniteDuration] = None)(
implicit
ec: ExecutionContext,
system: ActorSystem[?])
@@ -120,7 +126,8 @@ private[r2dbc] class DurableStateDao(settings:
StateSettings, connectionFactory:
protected lazy val transactionTimestampSql: String =
"transaction_timestamp()"
private val persistenceExt = Persistence(system)
- private val r2dbcExecutor = new R2dbcExecutor(connectionFactory, log,
settings.logDbCallsExceeding)(ec, system)
+ private val r2dbcExecutor =
+ new R2dbcExecutor(connectionFactory, log, settings.logDbCallsExceeding,
closeCallsExceeding)(ec, system)
protected val stateTable = settings.durableStateTableWithSchema
protected implicit val statePayloadCodec: PayloadCodec =
settings.durableStatePayloadCodec
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/r2dbc/state/scaladsl/mysql/MySQLDurableStateDao.scala
b/core/src/main/scala/org/apache/pekko/persistence/r2dbc/state/scaladsl/mysql/MySQLDurableStateDao.scala
index 0656dd1..d481879 100644
---
a/core/src/main/scala/org/apache/pekko/persistence/r2dbc/state/scaladsl/mysql/MySQLDurableStateDao.scala
+++
b/core/src/main/scala/org/apache/pekko/persistence/r2dbc/state/scaladsl/mysql/MySQLDurableStateDao.scala
@@ -41,8 +41,10 @@ import io.r2dbc.spi.Statement
@InternalApi
private[r2dbc] class MySQLDurableStateDao(
settings: StateSettings,
- connectionFactory: ConnectionFactory
-)(implicit ec: ExecutionContext, system: ActorSystem[?]) extends
DurableStateDao(settings, connectionFactory) {
+ connectionFactory: ConnectionFactory,
+ closeCallsExceeding: Option[FiniteDuration] = None
+)(implicit ec: ExecutionContext, system: ActorSystem[?])
+ extends DurableStateDao(settings, connectionFactory, closeCallsExceeding) {
MySQLJournalDao.settingRequirements(settings)
override lazy val transactionTimestampSql: String = "NOW(6)"
diff --git
a/core/src/test/scala/org/apache/pekko/persistence/r2dbc/TestDbLifecycle.scala
b/core/src/test/scala/org/apache/pekko/persistence/r2dbc/TestDbLifecycle.scala
index bbbf155..b6a6e30 100644
---
a/core/src/test/scala/org/apache/pekko/persistence/r2dbc/TestDbLifecycle.scala
+++
b/core/src/test/scala/org/apache/pekko/persistence/r2dbc/TestDbLifecycle.scala
@@ -18,6 +18,7 @@ import scala.concurrent.duration._
import org.apache.pekko
import pekko.actor.typed.ActorSystem
import pekko.persistence.Persistence
+import pekko.persistence.r2dbc.ConnectionFactorySettings
import pekko.persistence.r2dbc.internal.R2dbcExecutor
import com.typesafe.config.Config
import io.r2dbc.spi.ConnectionFactory
@@ -48,11 +49,15 @@ trait TestDbLifecycle extends BeforeAndAfterAll { this:
Suite =>
config.getConfig("pekko.persistence.r2dbc.connection-factory").atPath("test.connection-factory"))
// this assumes that journal, snapshot store and state use same connection
settings
- lazy val r2dbcExecutor: R2dbcExecutor =
+ lazy val r2dbcExecutor: R2dbcExecutor = {
+ val closeCallsExceeding =
+
ConnectionFactorySettings.closeCallsExceeding(config.getConfig("pekko.persistence.r2dbc.connection-factory"))
new R2dbcExecutor(
connectionFactoryProvider,
LoggerFactory.getLogger(getClass),
- journalSettings.logDbCallsExceeding)(typedSystem.executionContext,
typedSystem)
+ journalSettings.logDbCallsExceeding,
+ closeCallsExceeding)(typedSystem.executionContext, typedSystem)
+ }
lazy val persistenceExt: Persistence = Persistence(typedSystem)
diff --git
a/core/src/test/scala/org/apache/pekko/persistence/r2dbc/internal/R2dbcExecutorSpec.scala
b/core/src/test/scala/org/apache/pekko/persistence/r2dbc/internal/R2dbcExecutorSpec.scala
new file mode 100644
index 0000000..6ca55a8
--- /dev/null
+++
b/core/src/test/scala/org/apache/pekko/persistence/r2dbc/internal/R2dbcExecutorSpec.scala
@@ -0,0 +1,131 @@
+/*
+ * 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. <https://www.lightbend.com>
+ */
+
+package org.apache.pekko.persistence.r2dbc.internal
+
+import scala.concurrent.Await
+import scala.concurrent.duration._
+
+import org.apache.pekko
+import pekko.actor.testkit.typed.scaladsl.LogCapturing
+import pekko.actor.testkit.typed.scaladsl.ScalaTestWithActorTestKit
+import pekko.actor.typed.ActorSystem
+import pekko.persistence.r2dbc.ConnectionFactorySettings
+import pekko.persistence.r2dbc.TestConfig
+import pekko.persistence.r2dbc.TestData
+import pekko.persistence.r2dbc.TestDbLifecycle
+import com.typesafe.config.Config
+import com.typesafe.config.ConfigFactory
+import io.r2dbc.spi.Connection
+import io.r2dbc.spi.R2dbcNonTransientResourceException
+import io.r2dbc.spi.Wrapped
+import org.scalatest.{ Outcome, Pending }
+import org.scalatest.wordspec.AnyWordSpecLike
+
+object R2dbcExecutorSpec {
+ val config: Config = ConfigFactory
+ .parseString("""
+ pekko.persistence.r2dbc.connection-factory {
+ initial-size = 1
+ max-size = 1
+ close-calls-exceeding = 3 seconds
+ }
+ """)
+ .withFallback(TestConfig.config)
+
+ val dialect = config.getString("pekko.persistence.r2dbc.dialect")
+}
+
+class R2dbcExecutorSpec
+ extends ScalaTestWithActorTestKit(R2dbcExecutorSpec.config)
+ with AnyWordSpecLike
+ with TestDbLifecycle
+ with TestData
+ with LogCapturing {
+
+ override def withFixture(test: NoArgTest): Outcome =
+ if (R2dbcExecutorSpec.dialect == "mysql") {
+ Pending // uses pg_sleep which is Postgres-specific
+ } else {
+ super.withFixture(test)
+ }
+
+ override def typedSystem: ActorSystem[?] = system
+ private val table = "r2dbc_executor_spec"
+
+ private val closeCallsExceeding =
+ ConnectionFactorySettings.closeCallsExceeding(
+
system.settings.config.getConfig("pekko.persistence.r2dbc.connection-factory"))
+
+ case class Row(col: String)
+
+ override protected def beforeAll(): Unit = {
+ super.beforeAll()
+
+ Await.result(
+ r2dbcExecutor.executeDdl(s"beforeAll create table $table")(
+ _.createStatement(s"create table if not exists $table (col text)")),
+ 20.seconds)
+
+ r2dbcExecutor.updateOne("test")(_.createStatement(s"delete from
$table")).futureValue
+ }
+
+ "R2dbcExecutor" should {
+
+ "close connection when no response from update" in {
+ @volatile var c1: Connection = null
+ @volatile var c2: Connection = null
+
+ val result = r2dbcExecutor.update("test") { connection =>
+ c1 = connection.asInstanceOf[Wrapped[Connection]].unwrap()
+ Vector(
+ connection.createStatement(s"insert into $table (col) values ('a')"),
+ connection.createStatement("select pg_sleep(4)"),
+ connection.createStatement(s"insert into $table (col) values ('b')"))
+ }
+
+ Thread.sleep(closeCallsExceeding.get.toMillis)
+
+ // The request will fail with PostgresConnectionClosedException
+ result.failed.futureValue shouldBe a[R2dbcNonTransientResourceException]
+
+ r2dbcExecutor
+ .selectOne("test")(
+ { connection =>
+ c2 = connection.asInstanceOf[Wrapped[Connection]].unwrap()
+ connection.createStatement(s"select col from $table where col =
'a'")
+ },
+ row => Row(row.get("col", classOf[String])))
+ .futureValue shouldBe None
+
+ r2dbcExecutor
+ .selectOne("test")(
+ _.createStatement(s"select col from $table where col = 'b'"),
+ row => Row(row.get("col", classOf[String])))
+ .futureValue shouldBe None
+
+ // it shouldn't reuse the connection
+ c1 should not be theSameInstanceAs(c2)
+ }
+
+ "close connection when no response from update with auto-commit" in {
+ val result = r2dbcExecutor.updateOne("test")(_.createStatement("select
pg_sleep(4)"))
+
+ Thread.sleep(closeCallsExceeding.get.toMillis)
+
+ // The request will fail with PostgresConnectionClosedException
+ result.failed.futureValue shouldBe a[R2dbcNonTransientResourceException]
+ }
+
+ }
+}
diff --git
a/migration/src/main/scala/org/apache/pekko/persistence/r2dbc/migration/MigrationTool.scala
b/migration/src/main/scala/org/apache/pekko/persistence/r2dbc/migration/MigrationTool.scala
index 2811d97..c5fd3bc 100644
---
a/migration/src/main/scala/org/apache/pekko/persistence/r2dbc/migration/MigrationTool.scala
+++
b/migration/src/main/scala/org/apache/pekko/persistence/r2dbc/migration/MigrationTool.scala
@@ -36,6 +36,7 @@ import
pekko.persistence.query.scaladsl.CurrentPersistenceIdsQuery
import pekko.persistence.query.scaladsl.ReadJournal
import pekko.persistence.query.{ EventEnvelope => ClassicEventEnvelope }
import pekko.persistence.r2dbc.ConnectionFactoryProvider
+import pekko.persistence.r2dbc.ConnectionFactorySettings
import pekko.persistence.r2dbc.JournalSettings
import pekko.persistence.r2dbc.SnapshotSettings
import pekko.persistence.r2dbc.journal.JournalDao
@@ -141,7 +142,11 @@ class MigrationTool(system: ActorSystem[?]) {
private lazy val sourceSnapshotStore =
Persistence(system).snapshotStoreFor(sourceSnapshotPluginId)
private[r2dbc] val migrationDao =
- new MigrationToolDao(targetJournalConnectionFactory,
targetJournalSettings.logDbCallsExceeding)
+ new MigrationToolDao(
+ targetJournalConnectionFactory,
+ targetJournalSettings.logDbCallsExceeding,
+ ConnectionFactorySettings.closeCallsExceeding(
+
system.settings.config.getConfig(targetJournalSettings.useConnectionFactory)))
private lazy val createProgressTable: Future[Done] =
migrationDao.createProgressTable()
diff --git
a/migration/src/main/scala/org/apache/pekko/persistence/r2dbc/migration/MigrationToolDao.scala
b/migration/src/main/scala/org/apache/pekko/persistence/r2dbc/migration/MigrationToolDao.scala
index 7882ee0..7ef0415 100644
---
a/migration/src/main/scala/org/apache/pekko/persistence/r2dbc/migration/MigrationToolDao.scala
+++
b/migration/src/main/scala/org/apache/pekko/persistence/r2dbc/migration/MigrationToolDao.scala
@@ -37,10 +37,12 @@ import io.r2dbc.spi.ConnectionFactory
*/
@InternalApi private[r2dbc] class MigrationToolDao(
connectionFactory: ConnectionFactory,
- logDbCallsExceeding: FiniteDuration)(implicit ec: ExecutionContext,
system: ActorSystem[?]) {
+ logDbCallsExceeding: FiniteDuration,
+ closeCallsExceeding: Option[FiniteDuration])(implicit ec:
ExecutionContext, system: ActorSystem[?]) {
import MigrationToolDao.CurrentProgress
- private val r2dbcExecutor = new R2dbcExecutor(connectionFactory, log,
logDbCallsExceeding)(ec, system)
+ private val r2dbcExecutor =
+ new R2dbcExecutor(connectionFactory, log, logDbCallsExceeding,
closeCallsExceeding)(ec, system)
def createProgressTable(): Future[Done] = {
r2dbcExecutor.executeDdl("create migration progress table") { connection =>
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
