ptrdom commented on code in PR #620:
URL: https://github.com/apache/pekko-projection/pull/620#discussion_r4377534413
##########
r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/R2dbcProjectionImpl.scala:
##########
@@ -716,9 +716,13 @@ private[projection] class R2dbcProjectionImpl[Offset,
Envelope](
extends RunningProjection
with RunningProjectionManagement[Offset] {
+ // the periodic deletes of old timestamp offsets are only running while
the projection is running
+ offsetStore.startDeleteTask()
+
private val streamDone = source.run()
Review Comment:
The task is scheduled before `source.run()`. If materialization throws
synchronously in the `R2dbcRunningProjection` constructor, the object is never
returned, so nothing can call `stop()`. The task then leaks the same way as
before the fix. Suggest moving `offsetStore.startDeleteTask()` after `private
val streamDone = source.run()`, or wrapping it so the task is cancelled if
`run()` throws.
##########
r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/R2dbcTimestampOffsetStoreSpec.scala:
##########
@@ -994,6 +997,41 @@ class R2dbcTimestampOffsetStoreSpec
}
}
+ "stop the periodic delete of old records when the projection is stopped"
in {
Review Comment:
This test is directional for `startDeleteTask`/`stopDeleteTask` on the
store. Its name, though, says "when the projection is stopped", and it never
goes through `R2dbcRunningProjection`/`ProjectionBehavior`. If the two lines in
`R2dbcProjectionImpl` were removed, this test would still pass. Only the
indirect effect on the existing delete tests would catch it, and those now call
`startDeleteTask()` themselves.
Suggest either renaming the test to say what it covers, or adding a small
`R2dbcProjectionSpec` case that spawns `ProjectionBehavior` with a short
`delete-interval` and a spy offset store (or checks that `deleteTask` is empty
after `Stop`).
##########
r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/R2dbcOffsetStore.scala:
##########
@@ -319,11 +320,32 @@ private[projection] class R2dbcOffsetStore(
// To avoid delete requests when no new offsets have been stored since
previous delete
private val idle = new AtomicBoolean(false)
- system.scheduler.scheduleWithFixedDelay(
- settings.deleteInterval,
- settings.deleteInterval,
- () => deleteOldTimestampOffsets(),
- system.executionContext)
+ // Scheduled deletes of old timestamp offsets are only running while the
projection is running, otherwise the
+ // scheduled task would keep this offset store (and everything it
references) alive for the lifetime of the
+ // ActorSystem, also for projections that have been stopped.
+ private val deleteTask = new AtomicReference(Option.empty[Cancellable])
+
+ /**
+ * Start the periodic deletion of old timestamp offsets. Called when the
projection is started. Idempotent, and can
+ * be called again after `stopDeleteTask`, such as when a stopped projection
is started again.
+ */
+ def startDeleteTask(): Unit = {
+ if (deleteTask.get().isEmpty) {
+ val task = system.scheduler.scheduleWithFixedDelay(
+ settings.deleteInterval,
+ settings.deleteInterval,
+ () => deleteOldTimestampOffsets(),
+ system.executionContext)
+ if (!deleteTask.compareAndSet(None, Some(task)))
+ task.cancel() // concurrent start, keep the already registered task
+ }
+ }
+
+ /**
+ * Cancel the periodic deletion of old timestamp offsets. Called when the
projection is stopped.
+ */
+ def stopDeleteTask(): Unit =
+ deleteTask.getAndSet(None).foreach(_.cancel())
Review Comment:
Nit: if `startDeleteTask` and `stopDeleteTask` run concurrently, the task
can end up registered after the stop. That can't happen with the current
single-actor call pattern, but the scaladoc could say "not safe to call
concurrently with `stopDeleteTask`".
##########
r2dbc-int-test/src/test/scala/org/apache/pekko/projection/r2dbc/R2dbcTimestampOffsetStoreSpec.scala:
##########
@@ -963,6 +965,7 @@ class R2dbcTimestampOffsetStoreSpec
.withDeleteInterval(JDuration.ofMillis(500))
import deleteSettings._
val offsetStore = createOffsetStore(projectionId, deleteSettings)
+ offsetStore.startDeleteTask()
Review Comment:
Nit: this starts the task and never stops it, so it keeps firing for the
rest of the suite. Add `offsetStore.stopDeleteTask()` at the end, or use
try/finally.
##########
r2dbc/src/main/scala/org/apache/pekko/projection/r2dbc/internal/R2dbcProjectionImpl.scala:
##########
@@ -716,9 +716,13 @@ private[projection] class R2dbcProjectionImpl[Offset,
Envelope](
extends RunningProjection
with RunningProjectionManagement[Offset] {
+ // the periodic deletes of old timestamp offsets are only running while
the projection is running
+ offsetStore.startDeleteTask()
+
private val streamDone = source.run()
override def stop(): Future[Done] = {
+ offsetStore.stopDeleteTask()
Review Comment:
The task is tied to `stop()`, not to the stream ending. If the restart
backoff is configured with a finite `max-restarts` (the default is `-1`) and
the stream gives up, `streamDone` completes. The actor stays alive, so deletes
keep firing for a dead projection until the actor stops. This is an edge case.
If you address it, don't just add `streamDone.onComplete(_ =>
offsetStore.stopDeleteTask())`. `stopDeleteTask()` cancels whatever task is
currently registered on the shared store, so a late callback from the previous
run could cancel the task started by the next `run()` after a
management-command restart. One way is for `startDeleteTask()` to return the
`Cancellable` (or a token) and have each `R2dbcRunningProjection` cancel only
its own. Otherwise a note that this is out of scope is fine.
--
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]
