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-jdbc.git
The following commit(s) were added to refs/heads/main by this push:
new 97f43b62 Complete the journal write queue when the plugin actor stops
(#609)
97f43b62 is described below
commit 97f43b6235e736393c31246f5335494341c33e06
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Sep 11 09:37:27 2026 +0100
Complete the journal write queue when the plugin actor stops (#609)
Motivation:
`BaseDao` materializes a stream in its constructor, fed by a
`BoundedSourceQueue`
that batches journal row writes. The queue is never completed, so the stream
stays materialized for as long as the actor system lives. A new dao, and
with it
a new stream, is created every time the journal plugin actor starts, so
every
restart of that actor strands the stream of the previous incarnation.
`JdbcAsyncWriteJournal.postStop` already closes the database it owns, but
left
the write queue running.
Modification:
Added `BaseDao.completeWriteQueue()` (internal API) and called it from
`JdbcAsyncWriteJournal.postStop` before the database is closed. Both the
default
and the legacy journal dao extend `BaseDao`, so both are covered.
Result:
The write stream of a journal dao terminates when the plugin actor that
owns it
stops.
Tests:
- sbt "core/testOnly
org.apache.pekko.persistence.jdbc.journal.dao.BaseDaoSpec
org.apache.pekko.persistence.jdbc.journal.JdbcAsyncWriteJournalShutdownTest" -
passed; the shutdown test fails without the postStop change
- sbt "core/test" - 292 passed, 1 failed: H2ScalaCurrentEventsByTagTest
"should complete without any gaps in case events are being persisted when the
query is executed", which fails the same way on an unmodified checkout of main
on this machine
- sbt "core/mimaReportBinaryIssues" - passed
- scalafmt --mode diff-ref=upstream/main - no changes
References:
None - found while reviewing the main sources for resource leaks
---
.../jdbc/journal/JdbcAsyncWriteJournal.scala | 7 +-
.../persistence/jdbc/journal/dao/BaseDao.scala | 12 +++
.../JdbcAsyncWriteJournalShutdownTest.scala | 78 +++++++++++++++++++
.../persistence/jdbc/journal/dao/BaseDaoSpec.scala | 91 ++++++++++++++++++++++
4 files changed, 187 insertions(+), 1 deletion(-)
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/JdbcAsyncWriteJournal.scala
b/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/JdbcAsyncWriteJournal.scala
index c3324afa..5df96014 100644
---
a/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/JdbcAsyncWriteJournal.scala
+++
b/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/JdbcAsyncWriteJournal.scala
@@ -21,7 +21,7 @@ import pekko.Done
import pekko.actor.{ ActorSystem, ExtendedActorSystem }
import pekko.persistence.jdbc.config.JournalConfig
import pekko.persistence.jdbc.journal.JdbcAsyncWriteJournal.{
InPlaceUpdateEvent, WriteFinished }
-import pekko.persistence.jdbc.journal.dao.{ JournalDao, JournalDaoWithUpdates }
+import pekko.persistence.jdbc.journal.dao.{ BaseDao, JournalDao,
JournalDaoWithUpdates }
import pekko.persistence.jdbc.db.{ SlickDatabase, SlickExtension }
import pekko.persistence.journal.AsyncWriteJournal
import pekko.persistence.{ AtomicWrite, PersistentRepr }
@@ -129,6 +129,11 @@ class JdbcAsyncWriteJournal(config: Config) extends
AsyncWriteJournal {
.map(_ => ())
override def postStop(): Unit = {
+ journalDao match {
+ // the write queue keeps a stream materialized, a new dao is created
every time this actor starts
+ case baseDao: BaseDao[?] => baseDao.completeWriteQueue()
+ case _ =>
+ }
if (slickDb.allowShutdown) {
// Since a (new) db is created when this actor (re)starts, we must close
it when the actor stops
db.close()
diff --git
a/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/BaseDao.scala
b/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/BaseDao.scala
index 4bd0114d..a140676b 100644
---
a/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/BaseDao.scala
+++
b/core/src/main/scala/org/apache/pekko/persistence/jdbc/journal/dao/BaseDao.scala
@@ -15,6 +15,7 @@
package org.apache.pekko.persistence.jdbc.journal.dao
import org.apache.pekko
+import pekko.annotation.InternalApi
import pekko.persistence.jdbc.config.BaseDaoConfig
import pekko.stream.scaladsl.{ Keep, Sink, Source }
import pekko.stream.{ BoundedSourceQueue, Materializer, QueueOfferResult }
@@ -59,4 +60,15 @@ abstract class BaseDao[T] {
}
}
+ /**
+ * INTERNAL API
+ *
+ * Completes the queue that batches the journal row writes, which terminates
the stream that was materialized for
+ * this dao once the batches that were already enqueued have been written. A
dao (and with it a new stream) is
+ * created every time the plugin actor starts, so a stream that is never
completed stays materialized until the
+ * actor system shuts down.
+ */
+ @InternalApi
+ private[jdbc] def completeWriteQueue(): Unit = writeQueue.complete()
+
}
diff --git
a/core/src/test/scala/org/apache/pekko/persistence/jdbc/journal/JdbcAsyncWriteJournalShutdownTest.scala
b/core/src/test/scala/org/apache/pekko/persistence/jdbc/journal/JdbcAsyncWriteJournalShutdownTest.scala
new file mode 100644
index 00000000..da2b0dac
--- /dev/null
+++
b/core/src/test/scala/org/apache/pekko/persistence/jdbc/journal/JdbcAsyncWriteJournalShutdownTest.scala
@@ -0,0 +1,78 @@
+/*
+ * 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.jdbc.journal
+
+import java.util.concurrent.CopyOnWriteArrayList
+
+import scala.collection.immutable.Seq
+import scala.concurrent.ExecutionContext
+import scala.jdk.CollectionConverters._
+
+import org.apache.pekko
+import pekko.persistence.Persistence
+import pekko.persistence.jdbc.SingleActorSystemPerTestSpec
+import pekko.persistence.jdbc.config.JournalConfig
+import pekko.persistence.jdbc.journal.dao.DefaultJournalDao
+import pekko.serialization.Serialization
+import pekko.stream.Materializer
+import com.typesafe.config.ConfigValueFactory
+import slick.jdbc.JdbcBackend.Database
+import slick.jdbc.JdbcProfile
+
+/**
+ * A journal dao that publishes the instances that the journal plugin creates,
so that a test can look at the state of
+ * the dao after the plugin actor has stopped.
+ */
+class RecordingJournalDao(
+ db: Database,
+ profile: JdbcProfile,
+ journalConfig: JournalConfig,
+ serialization: Serialization)(implicit ec: ExecutionContext, mat:
Materializer)
+ extends DefaultJournalDao(db, profile, journalConfig, serialization) {
+ RecordingJournalDao.instances.add(this)
+}
+
+object RecordingJournalDao {
+ val instances = new CopyOnWriteArrayList[RecordingJournalDao]
+
+ def clear(): Unit = instances.clear()
+}
+
+class JdbcAsyncWriteJournalShutdownTest
+ extends SingleActorSystemPerTestSpec(
+ "h2-shared-db-application.conf",
+ Map("jdbc-journal.dao" ->
ConfigValueFactory.fromAnyRef(classOf[RecordingJournalDao].getName))) {
+
+ it should "complete the write queue of its dao when the journal actor stops"
in {
+ RecordingJournalDao.clear()
+ withActorSystem { implicit system =>
+ val journal = Persistence(system).journalFor("jdbc-journal")
+ // the plugin actor, and with it the dao, is created asynchronously
+ val dao = eventually {
+ val daos = RecordingJournalDao.instances.asScala.toList
+ daos should have size 1
+ daos.head
+ }
+
+ killActors(journal)
+
+ val failure = dao.queueWriteJournalRows(Seq.empty).failed.futureValue
+ failure.getMessage should include("the queue was closed")
+ }
+ }
+}
diff --git
a/core/src/test/scala/org/apache/pekko/persistence/jdbc/journal/dao/BaseDaoSpec.scala
b/core/src/test/scala/org/apache/pekko/persistence/jdbc/journal/dao/BaseDaoSpec.scala
new file mode 100644
index 00000000..9dfd1cb4
--- /dev/null
+++
b/core/src/test/scala/org/apache/pekko/persistence/jdbc/journal/dao/BaseDaoSpec.scala
@@ -0,0 +1,91 @@
+/*
+ * 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.jdbc.journal.dao
+
+import java.util.concurrent.ConcurrentLinkedQueue
+
+import scala.collection.immutable.Seq
+import scala.concurrent.{ ExecutionContext, Future }
+import scala.concurrent.duration._
+import scala.jdk.CollectionConverters._
+
+import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.persistence.jdbc.config.BaseDaoConfig
+import pekko.stream.{ Materializer, SystemMaterializer }
+import pekko.testkit.TestKit
+import com.typesafe.config.ConfigFactory
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.concurrent.ScalaFutures
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpecLike
+
+object BaseDaoSpec {
+ class TestDao(override val baseDaoConfig: BaseDaoConfig)(
+ implicit val ec: ExecutionContext,
+ val mat: Materializer)
+ extends BaseDao[String] {
+
+ val written = new ConcurrentLinkedQueue[String]
+
+ override def writeJournalRows(xs: Seq[String]): Future[Unit] = {
+ xs.foreach(written.add)
+ Future.unit
+ }
+ }
+}
+
+class BaseDaoSpec extends TestKit(ActorSystem("BaseDaoSpec")) with
AnyWordSpecLike with Matchers with ScalaFutures
+ with BeforeAndAfterAll {
+ import BaseDaoSpec._
+
+ implicit val defaultPatience: PatienceConfig = PatienceConfig(timeout =
1.minute)
+ implicit val ec: ExecutionContext = system.dispatcher
+ implicit val mat: Materializer = SystemMaterializer(system).materializer
+
+ private val daoConfig = new BaseDaoConfig(
+ ConfigFactory.parseString("""
+ bufferSize = 100
+ batchSize = 10
+ replayBatchSize = 10
+ parallelism = 1
+ """))
+
+ override def afterAll(): Unit = TestKit.shutdownActorSystem(system)
+
+ "BaseDao" must {
+ "write the rows that are queued" in {
+ val dao = new TestDao(daoConfig)
+
+ dao.queueWriteJournalRows(Seq("a", "b")).futureValue
+
+ dao.written.asScala.toList shouldBe List("a", "b")
+ }
+
+ "reject rows once the write queue has been completed" in {
+ val dao = new TestDao(daoConfig)
+ dao.queueWriteJournalRows(Seq("a")).futureValue
+
+ dao.completeWriteQueue()
+
+ val failure = dao.queueWriteJournalRows(Seq("b")).failed.futureValue
+ failure.getMessage should include("the queue was closed")
+ dao.written.asScala.toList shouldBe List("a")
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
