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: [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]
