pjfanning commented on code in PR #620:
URL: https://github.com/apache/pekko-projection/pull/620#discussion_r4379547458
##########
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:
Did both in 6ee59a6c. The store test is renamed to "stop the periodic delete
of old records when the scheduled task is cancelled".
`R2dbcTimestampOffsetProjectionSpec` has a new case, "only run the periodic
delete while the projection is running". It spawns `ProjectionBehavior` with a
spy offset store that counts the scheduled deletes, and checks that there are
none before spawn, that deletes run while running and after a
`ProjectionManagement.updateOffset` restart, and that there are none after
`Stop`. It fails when the cancels in `R2dbcRunningProjection` are disabled.
##########
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:
Fixed in 6ee59a6c. The test now cancels the task in a `try/finally`.
##########
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:
No longer applies after 6ee59a6c. `startDeleteTask`/`stopDeleteTask` are
replaced by a stateless `scheduleDeleteTask(): Cancellable`, and each running
projection owns the task it scheduled, so there is no shared registration to
race on.
--
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]
