pjfanning opened a new pull request, #619:
URL: https://github.com/apache/pekko-projection/pull/619
### Motivation
`FilterStage` materialises a `currentEventsByPersistenceIdTyped` query per
replay with `Sink.queue()` and keeps it in `replayInProgress`:
```scala
val queue = currentEventsByPersistenceIdQuery
.currentEventsByPersistenceIdTyped[Any](pid, fromSeqNr, Long.MaxValue)
.runWith(Sink.queue())(materializer)
replayInProgress = replayInProgress.updated(pid, ReplaySession(fromSeqNr,
queue))
```
`queue.cancel()` is only called when a replay is replaced by a newer replay
for the same persistenceId. There is no `postStop`, so when the stage completes
or fails (`failStage` from the replay callback, or the consumer cancelling the
gRPC stream) every replay still in progress - up to `replayParallelism` per
stream - stays materialised with an un-pulled `SinkQueue`, keeping the query
stream and the resources it holds, such as a database connection, alive
indefinitely.
### Modification
Cancel the queues of all replays still in `replayInProgress` from `postStop`.
### Result
Replay streams are cancelled when the stage stops, instead of being leaked.
### Tests
- New `FilterStageSpec` case "cancel replay streams that are still in
progress when stopped": starts a replay backed by a source that never emits,
stops the stage, and expects the replay source to be cancelled. It fails
without the fix (the source is never cancelled).
- `sbt "grpc-test/Test/testOnly *FilterStageSpec"` - 9 passed
- `sbt checkCodeStyle` - success
- MiMa not run - `MimaPlugin` is disabled for the `grpc` module in
`build.sbt`; the change only adds a method to an anonymous `GraphStageLogic`
inside `@InternalApi` `FilterStage`.
### References
None - found during a resource-leak audit of the main sources
--
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]
