only run R2dbcOffsetStore delete task while the projection is running

[pjfanning]

Motivation

R2dbcOffsetStore scheduled deleteOldTimestampOffsets with a fixed delay in its constructor and discarded the returned Cancellable:

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