You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@pekko.apache.org by "pjfanning (via GitHub)" <gi...@apache.org> on 2026/09/06 22:08:25 UTC
[PR] Complete the journal write queue when the plugin actor stops [pekko-persistence-jdbc]
pjfanning opened a new pull request, #609:
URL: https://github.com/apache/pekko-persistence-jdbc/pull/609
### Motivation
`BaseDao` materializes a stream in its constructor, fed by a `BoundedSourceQueue` that batches journal row writes:
```scala
private val writeQueue: BoundedSourceQueue[(Promise[Unit], Seq[T])] = Source
.queue[(Promise[Unit], Seq[T])](baseDaoConfig.bufferSize)
...
.toMat(Sink.ignore)(Keep.left)
.run()
```
Nothing ever completes that queue, 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. `JdbcAsyncWriteJournalShutdownTest` stops the real plugin actor and asserts that its dao rejects further rows with "the queue was closed"; it 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". That test fails the same way on an unmodified checkout of `main` on this machine, so it is pre-existing flakiness and unrelated.
- `sbt "core/mimaReportBinaryIssues"` - passed
- `scalafmt --mode diff-ref=upstream/main` - no changes
### References
None - found while reviewing the main sources for resource leaks
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [PR] Complete the journal write queue when the plugin actor stops [pekko-persistence-jdbc]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
pjfanning merged PR #609:
URL: https://github.com/apache/pekko-persistence-jdbc/pull/609
--
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: notifications-unsubscribe@pekko.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org