feat: Make it possible to disable offset deletion (#1195)

akka · Sep 12, 2024 · b6418a0 · b6418a0
1 parent 15c944d
commit b6418a0
Show file tree

Hide file tree

Showing 3 changed files with 13 additions and 7 deletions.
diff --git a/akka-projection-r2dbc/src/main/resources/reference.conf b/akka-projection-r2dbc/src/main/resources/reference.conf
@@ -30,7 +30,7 @@ akka.projection.r2dbc {
     evict-interval = 10 seconds
 
     # Remove old entries outside the time-window from the offset store database
-    # with this frequency.
+    # with this frequency. Can be disabled with `off`.
     delete-interval = 1 minute
 
     # Trying to batch insert offsets in batches of this size.

diff --git a/akka-projection-r2dbc/src/main/scala/akka/projection/r2dbc/R2dbcProjectionSettings.scala b/akka-projection-r2dbc/src/main/scala/akka/projection/r2dbc/R2dbcProjectionSettings.scala
@@ -39,6 +39,11 @@ object R2dbcProjectionSettings {
         case _     => config.getDuration("log-db-calls-exceeding").asScala
       }
 
+    val deleteInterval = config.getString("offset-store.delete-interval").toLowerCase(Locale.ROOT) match {
+      case "off" => JDuration.ZERO
+      case _     => config.getDuration("offset-store.delete-interval")
+    }
+
     new R2dbcProjectionSettings(
       schema = Option(config.getString("offset-store.schema")).filterNot(_.trim.isEmpty),
       offsetTable = config.getString("offset-store.offset-table"),
@@ -48,7 +53,7 @@ object R2dbcProjectionSettings {
       timeWindow = config.getDuration("offset-store.time-window"),
       keepNumberOfEntries = config.getInt("offset-store.keep-number-of-entries"),
       evictInterval = config.getDuration("offset-store.evict-interval"),
-      deleteInterval = config.getDuration("offset-store.delete-interval"),
+      deleteInterval,
       logDbCallsExceeding,
       warnAboutFilteredEventsInFlow = config.getBoolean("warn-about-filtered-events-in-flow"),
       offsetBatchSize = config.getInt("offset-store.offset-batch-size"))

diff --git a/akka-projection-r2dbc/src/main/scala/akka/projection/r2dbc/internal/R2dbcOffsetStore.scala b/akka-projection-r2dbc/src/main/scala/akka/projection/r2dbc/internal/R2dbcOffsetStore.scala
@@ -243,11 +243,12 @@ private[projection] class R2dbcOffsetStore(
   // To trigger next deletion after in-memory eviction
   private val triggerDeletion = new AtomicBoolean(false)
 
-  system.scheduler.scheduleWithFixedDelay(
-    settings.deleteInterval,
-    settings.deleteInterval,
-    () => deleteOldTimestampOffsets(),
-    system.executionContext)
+  if (!settings.deleteInterval.isZero && !settings.deleteInterval.isNegative)
+    system.scheduler.scheduleWithFixedDelay(
+      settings.deleteInterval,
+      settings.deleteInterval,
+      () => deleteOldTimestampOffsets(),
+      system.executionContext)
 
   private def timestampOf(persistenceId: String, sequenceNr: Long): Future[Option[Instant]] = {
     sourceProvider match {