chore: start stats saving loop on init of sync persistence. (#12968)

airbytehq · Jun 27, 2024 · c212200 · c212200
1 parent bff305d
commit c212200
Showing 1 changed file with 17 additions and 9 deletions.
diff --git a/...ons-worker/src/main/kotlin/io/airbyte/workers/internal/syncpersistence/SyncPersistence.kt b/...ons-worker/src/main/kotlin/io/airbyte/workers/internal/syncpersistence/SyncPersistence.kt
@@ -115,6 +115,12 @@ class SyncPersistenceImpl
       this.retryWithJitterConfig = retryWithJitterConfig
     }
 
+    init {
+      if (isFrequencyOverrideEnabled()) {
+        startBackgroundFlushStateTask(connectionId)
+      }
+    }
+
     @Trace
     override fun persist(
       connectionId: UUID,
@@ -277,25 +283,27 @@ class SyncPersistenceImpl
 
       val haveStateToFlush = stateToFlush?.isEmpty() == false
 
-      val frequencyOverride =
-        featureFlagClient.intVariation(
-          SyncStatsFlushPeriodOverrideSeconds,
-          Multi(listOf(Workspace(workspaceId), Connection(connectionId))),
-        )
-
-      val frequencyOverrideOn = frequencyOverride > -1
-
       // We prepare stats to commit. We generate the payload here to keep track as close as possible to
       // the states that are going to be persisted. There is a minute race chance.
       // We also only want to generate the stats payload when roll-over state buffers. This is to avoid
       // updating the committed data counters ahead of the states because this counter is currently
       // decoupled from the state persistence.
       // This design favoring accuracy of committed data counters over freshness of emitted data counters.
-      if (haveStateToFlush || frequencyOverrideOn) {
+      if (haveStateToFlush || isFrequencyOverrideEnabled()) {
         statsToPersist = buildSaveStatsRequest(syncStatsTracker, jobId, attemptNumber, connectionId)
       }
     }
 
+    private fun isFrequencyOverrideEnabled(): Boolean {
+      val frequencyOverride =
+        featureFlagClient.intVariation(
+          SyncStatsFlushPeriodOverrideSeconds,
+          Multi(listOf(Workspace(workspaceId), Connection(connectionId))),
+        )
+
+      return frequencyOverride > -1
+    }
+
     private fun doFlushState() {
       if (stateToFlush == null || stateToFlush?.isEmpty() == true) {
         return