pjfanning opened a new pull request, #620:
URL: https://github.com/apache/pekko-projection/pull/620
### Motivation
`R2dbcOffsetStore` scheduled `deleteOldTimestampOffsets` with a fixed delay
in its constructor and discarded the returned `Cancellable`:
```scala
system.scheduler.scheduleWithFixedDelay(
settings.deleteInterval, settings.deleteInterval,
() => deleteOldTimestampOffsets(), system.executionContext)
```
The store is created eagerly by every `R2dbcProjection` factory method
(`R2dbcProjectionImpl.createOffsetStore`), so the task starts before the
projection runs, and nothing ever cancels it. It keeps firing every
`delete-interval` (`1 minute` by default) for the lifetime of the
`ActorSystem`, also after the projection has been stopped, and it keeps the
offset store, its source provider and its `R2dbcExecutor` reachable, so each
stopped projection is retained. This adds up when projections are stopped and
started, such as on ShardedDaemonProcess rebalance.
### Modification
Keep the `Cancellable` and expose `startDeleteTask()`/`stopDeleteTask()` on
the offset store. `R2dbcRunningProjection` starts the task when it is created
and cancels it in `stop()`, so the deletes run while the projection is running,
and a projection that is stopped and started again (`ProjectionBehavior` does
this for the offset management commands) gets the task back.
### Result
No scheduled task, and no retained offset store, for a projection that has
been stopped or that was never run.
One behaviour note: the deletes are not scheduled for the
`ProjectionTestKit` path, which runs `mappedSource()` without a
`RunningProjection`. Tests that exercise deletion call the offset store
directly, and `R2dbcTimestampOffsetStoreSpec` "periodically delete old records"
now starts the task explicitly.
### Tests
- New `R2dbcTimestampOffsetStoreSpec` case "stop the periodic delete of old
records when the projection is stopped": counts the scheduled delete ticks
through a probe, asserts they stop after `stopDeleteTask()` and resume after a
later `startDeleteTask()`. It fails if `stopDeleteTask()` does not cancel
(verified by neutering the cancel).
- `sbt "r2dbc-int-test/Test/testOnly *R2dbcTimestampOffsetStoreSpec"` - 21
passed
- `sbt "r2dbc-int-test/test"` - 116 passed. `RuntimePluginConfigSpec` aborts
in my environment with `database "database1" does not exist`; it aborts
identically on the unmodified main branch, so it is unrelated to this change.
- `sbt "r2dbc/mimaReportBinaryIssues"` - success
- `sbt checkCodeStyle` - success
### 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]
