delete feature flag used for tracking committed records for global st…

…ate (#10560)

airbyte-commons-worker/src/main/java/io/airbyte/workers/general/ReplicationWorkerFactory.java

@@ -28,7 +28,6 @@
 import io.airbyte.featureflag.Source;
 import io.airbyte.featureflag.SourceDefinition;
 import io.airbyte.featureflag.SourceType;
-import io.airbyte.featureflag.TrackCommittedStatsWhenUsingGlobalState;
 import io.airbyte.featureflag.Workspace;
 import io.airbyte.metrics.lib.MetricAttribute;
 import io.airbyte.metrics.lib.MetricClient;
@@ -168,7 +167,7 @@ public ReplicationWorker create(final ReplicationInput replicationInput,
 
     log.info("Setting up replication worker...");
     final SyncPersistence syncPersistence = createSyncPersistence(syncPersistenceFactory, replicationInput, sourceLauncherConfig);
-    final AirbyteMessageTracker messageTracker = createMessageTracker(featureFlagClient, syncPersistence, featureFlags, replicationInput);
+    final AirbyteMessageTracker messageTracker = createMessageTracker(syncPersistence, featureFlags, replicationInput);
 
     return createReplicationWorker(airbyteSource, airbyteDestination, messageTracker,
         syncPersistence, recordSchemaValidator, fieldSelector, heartbeatTimeoutChaperone,
@@ -260,14 +259,11 @@ private static DestinationTimeoutMonitor createDestinationTimeout(final FeatureF
   /**
    * Create MessageTracker.
    */
-  private static AirbyteMessageTracker createMessageTracker(final FeatureFlagClient featureFlagClient,
-                                                            final SyncPersistence syncPersistence,
+  private static AirbyteMessageTracker createMessageTracker(final SyncPersistence syncPersistence,
                                                             final FeatureFlags featureFlags,
                                                             final ReplicationInput replicationInput) {
-    Context context = new Multi(List.of(new Workspace(replicationInput.getWorkspaceId()), new Connection(replicationInput.getConnectionId())));
-    boolean trackCommittedStatsWhenUsingGlobalState = featureFlagClient.boolVariation(TrackCommittedStatsWhenUsingGlobalState.INSTANCE, context);
     return new AirbyteMessageTracker(syncPersistence, featureFlags, replicationInput.getSourceLauncherConfig().getDockerImage(),
-        replicationInput.getDestinationLauncherConfig().getDockerImage(), trackCommittedStatsWhenUsingGlobalState);
+        replicationInput.getDestinationLauncherConfig().getDockerImage());
   }
 
   /**

airbyte-commons-worker/src/main/kotlin/io/airbyte/workers/internal/bookkeeping/AirbyteMessageTracker.kt

@@ -24,7 +24,6 @@ class AirbyteMessageTracker(
   featureFlags: FeatureFlags,
   private val sourceDockerImage: String,
   private val destinationDockerImage: String,
-  private val trackCommittedStatsWhenUsingGlobalState: Boolean,
 ) {
   private val dstErrorTraceMsgs = ArrayList<AirbyteTraceMessage>()
   private val srcErrorTraceMsgs = ArrayList<AirbyteTraceMessage>()
@@ -43,7 +42,7 @@ class AirbyteMessageTracker(
     when (msg.type) {
       AirbyteMessage.Type.TRACE -> handleEmittedTrace(msg.trace, AirbyteMessageOrigin.SOURCE)
       AirbyteMessage.Type.RECORD -> syncStatsTracker.updateStats(msg.record)
-      AirbyteMessage.Type.STATE -> syncStatsTracker.updateSourceStatesStats(msg.state, trackCommittedStatsWhenUsingGlobalState)
+      AirbyteMessage.Type.STATE -> syncStatsTracker.updateSourceStatesStats(msg.state)
       AirbyteMessage.Type.CONTROL -> logger.debug { "Control message not currently tracked." }
       else -> logger.warn { "Invalid message type for message: $msg" }
     }
@@ -63,7 +62,7 @@ class AirbyteMessageTracker(
       AirbyteMessage.Type.STATE ->
         msg.state?.let {
           stateAggregator.ingest(it)
-          syncStatsTracker.updateDestinationStateStats(it, trackCommittedStatsWhenUsingGlobalState)
+          syncStatsTracker.updateDestinationStateStats(it)
         }
       AirbyteMessage.Type.CONTROL -> logger.debug { "Control message not currently tracked." }
       else -> logger.warn { " Invalid message type for message: $msg" }

airbyte-commons-worker/src/main/kotlin/io/airbyte/workers/internal/bookkeeping/ParallelStreamStatsTracker.kt

@@ -61,49 +61,33 @@ class ParallelStreamStatsTracker(private val metricClient: MetricClient) : SyncS
     }
   }
 
-  override fun updateSourceStatesStats(
-    stateMessage: AirbyteStateMessage,
-    trackCommittedStatsWhenUsingGlobalState: Boolean,
-  ) {
-    if (trackCommittedStatsWhenUsingGlobalState) {
-      when (stateMessage.type) {
-        AirbyteStateMessage.AirbyteStateType.GLOBAL -> {
-          stateMessage.global.streamStates.forEach {
-            getOrCreateStreamStatsTracker(getNameNamespacePair(it.streamDescriptor))
-              .trackStateFromSource(stateMessage)
-          }
-        }
-        else -> {
-          getOrCreateStreamStatsTracker(getNameNamespacePair(stateMessage))
+  override fun updateSourceStatesStats(stateMessage: AirbyteStateMessage) {
+    when (stateMessage.type) {
+      AirbyteStateMessage.AirbyteStateType.GLOBAL -> {
+        stateMessage.global.streamStates.forEach {
+          getOrCreateStreamStatsTracker(getNameNamespacePair(it.streamDescriptor))
             .trackStateFromSource(stateMessage)
         }
       }
-    } else {
-      getOrCreateStreamStatsTracker(getNameNamespacePair(stateMessage))
-        .trackStateFromSource(stateMessage)
+      else -> {
+        getOrCreateStreamStatsTracker(getNameNamespacePair(stateMessage))
+          .trackStateFromSource(stateMessage)
+      }
     }
   }
 
-  override fun updateDestinationStateStats(
-    stateMessage: AirbyteStateMessage,
-    trackCommittedStatsWhenUsingGlobalState: Boolean,
-  ) {
-    if (trackCommittedStatsWhenUsingGlobalState) {
-      when (stateMessage.type) {
-        AirbyteStateMessage.AirbyteStateType.GLOBAL -> {
-          stateMessage.global.streamStates.forEach {
-            getOrCreateStreamStatsTracker(getNameNamespacePair(it.streamDescriptor))
-              .trackStateFromDestination(stateMessage)
-          }
-        }
-        else -> {
-          getOrCreateStreamStatsTracker(getNameNamespacePair(stateMessage))
+  override fun updateDestinationStateStats(stateMessage: AirbyteStateMessage) {
+    when (stateMessage.type) {
+      AirbyteStateMessage.AirbyteStateType.GLOBAL -> {
+        stateMessage.global.streamStates.forEach {
+          getOrCreateStreamStatsTracker(getNameNamespacePair(it.streamDescriptor))
             .trackStateFromDestination(stateMessage)
         }
       }
-    } else {
-      getOrCreateStreamStatsTracker(getNameNamespacePair(stateMessage))
-        .trackStateFromDestination(stateMessage)
+      else -> {
+        getOrCreateStreamStatsTracker(getNameNamespacePair(stateMessage))
+          .trackStateFromDestination(stateMessage)
+      }
     }
   }
 

airbyte-commons-worker/src/main/kotlin/io/airbyte/workers/internal/bookkeeping/SyncStatsTracker.kt

@@ -32,18 +32,12 @@ interface SyncStatsTracker {
   /**
    * Update the stats count from the source state message.
    */
-  fun updateSourceStatesStats(
-    stateMessage: AirbyteStateMessage,
-    trackCommittedStatsWhenUsingGlobalState: Boolean,
-  )
+  fun updateSourceStatesStats(stateMessage: AirbyteStateMessage)
 
   /**
    * Update the stats count from the source state message.
    */
-  fun updateDestinationStateStats(
-    stateMessage: AirbyteStateMessage,
-    trackCommittedStatsWhenUsingGlobalState: Boolean,
-  )
+  fun updateDestinationStateStats(stateMessage: AirbyteStateMessage)
 
   /**
    * Get the per-stream committed bytes count.

airbyte-commons-worker/src/main/kotlin/io/airbyte/workers/internal/syncpersistence/SyncPersistence.kt

@@ -398,20 +398,14 @@ class SyncPersistenceImpl
       syncStatsTracker.updateEstimates(estimate)
     }
 
-    override fun updateSourceStatesStats(
-      stateMessage: AirbyteStateMessage,
-      trackCommittedStatsWhenUsingGlobalState: Boolean,
-    ) {
+    override fun updateSourceStatesStats(stateMessage: AirbyteStateMessage) {
       isReceivingStats = true
-      syncStatsTracker.updateSourceStatesStats(stateMessage, trackCommittedStatsWhenUsingGlobalState)
+      syncStatsTracker.updateSourceStatesStats(stateMessage)
     }
 
-    override fun updateDestinationStateStats(
-      stateMessage: AirbyteStateMessage,
-      trackCommittedStatsWhenUsingGlobalState: Boolean,
-    ) {
+    override fun updateDestinationStateStats(stateMessage: AirbyteStateMessage) {
       isReceivingStats = true
-      syncStatsTracker.updateDestinationStateStats(stateMessage, trackCommittedStatsWhenUsingGlobalState)
+      syncStatsTracker.updateDestinationStateStats(stateMessage)
     }
   }
 

airbyte-commons-worker/src/test/java/io/airbyte/workers/internal/bookkeeping/AirbyteMessageTrackerTest.java

@@ -34,7 +34,7 @@ class AirbyteMessageTrackerTest {
   @BeforeEach
   void setup() {
     this.messageTracker =
-        new AirbyteMessageTracker(syncStatsTracker, new EnvVariableFeatureFlags(), "airbyte/source-image", "airbyte/destination-image", false);
+        new AirbyteMessageTracker(syncStatsTracker, new EnvVariableFeatureFlags(), "airbyte/source-image", "airbyte/destination-image");
   }
 
   @Test
@@ -88,7 +88,7 @@ void testAcceptFromSourceState() {
 
     messageTracker.acceptFromSource(state);
 
-    verify(syncStatsTracker).updateSourceStatesStats(state.getState(), false);
+    verify(syncStatsTracker).updateSourceStatesStats(state.getState());
   }
 
   @Test
@@ -169,7 +169,7 @@ void testAcceptFromDestinationState() {
 
     messageTracker.acceptFromDestination(state);
 
-    verify(syncStatsTracker).updateDestinationStateStats(state.getState(), false);
+    verify(syncStatsTracker).updateDestinationStateStats(state.getState());
   }
 
   @Test