pjfanning commented on code in PR #620:
URL: https://github.com/apache/pekko-projection/pull/620#discussion_r4379546691
##########
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:
Fixed in 6ee59a6c. The task is now scheduled after `private val streamDone =
source.run()`, so nothing is scheduled if materialization throws.
##########
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:
Fixed in 6ee59a6c, using the approach you suggested.
`R2dbcOffsetStore.scheduleDeleteTask()` returns the `Cancellable` and the store
keeps no task state. Each `R2dbcRunningProjection` cancels only its own task,
in `stop()` and in `streamDone.onComplete`. A late completion of a previous run
therefore cannot cancel the task of the next run. Covered by the new
`R2dbcTimestampOffsetProjectionSpec` case "stop the periodic delete when the
projection gives up after max restarts", which fails without the completion
callback.
--
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]
