Remove output state peristence. Add catalog validation to failure cas…

…e. (#12102)
airbytehq · Apr 12, 2024 · b702635 · b702635
1 parent c322b18
commit b702635
Show file tree

Hide file tree

Showing 4 changed files with 12 additions and 37 deletions.
diff --git a/airbyte-config/config-models/src/main/resources/types/StandardSyncOutput.yaml b/airbyte-config/config-models/src/main/resources/types/StandardSyncOutput.yaml
@@ -26,5 +26,3 @@ properties:
     "$ref": ActivityPayloadURI.yaml
   catalogUri:
     "$ref": ActivityPayloadURI.yaml
-  stateUri:
-    "$ref": ActivityPayloadURI.yaml
diff --git a/airbyte-featureflag/src/main/kotlin/FlagDefinitions.kt b/airbyte-featureflag/src/main/kotlin/FlagDefinitions.kt
@@ -202,12 +202,8 @@ object WriteReplicationOutputToObjectStorage : Temporary<Boolean>(key = "platfor
 
 object ReadReplicationOutputFromObjectStorage : Temporary<Boolean>(key = "platform.read-replication-output-from-object-storage", default = false)
 
-object WriteOutputStateToObjectStorage : Temporary<Boolean>(key = "platform.write-output-state-to-object-storage", default = false)
-
 object WriteOutputCatalogToObjectStorage : Temporary<Boolean>(key = "platform.write-output-catalog-to-object-storage", default = false)
 
-object NullOutputStateOnSyncOutput : Temporary<Boolean>(key = "platform.null-output-state-on-sync-output", default = false)
-
 object NullOutputCatalogOnSyncOutput : Temporary<Boolean>(key = "platform.null-output-catalog-on-sync-output", default = false)
 
 object UseCustomK8sInitCheck : Temporary<Boolean>(key = "platform.use-custom-k8s-init-check", default = true)

diff --git a/...irbyte/workers/temporal/scheduling/activities/JobCreationAndStatusUpdateActivityImpl.java b/...irbyte/workers/temporal/scheduling/activities/JobCreationAndStatusUpdateActivityImpl.java
@@ -105,20 +105,11 @@ public void jobSuccessWithAttemptNumber(final JobSuccessInputWithAttemptNumber i
         input.getJobId(),
         input.getAttemptNumber());
 
-    if (output != null) {
-      if (output.getState() != null && output.getStateUri() != null) {
-        stateClient.validate(
-            output.getState(),
-            output.getStateUri(),
-            new ArrayList<>());
-      }
-
-      if (output.getOutputCatalog() != null && output.getCatalogUri() != null) {
-        catalogClient.validate(
-            output.getOutputCatalog(),
-            output.getCatalogUri(),
-            new ArrayList<>());
-      }
+    if (output != null && output.getOutputCatalog() != null && output.getCatalogUri() != null) {
+      catalogClient.validate(
+          output.getOutputCatalog(),
+          output.getCatalogUri(),
+          new ArrayList<>());
     }
 
     try {
@@ -164,6 +155,13 @@ public void attemptFailureWithAttemptNumber(final AttemptNumberFailureInput inpu
         input.getJobId(),
         input.getAttemptNumber());
 
+    if (output != null && output.getOutputCatalog() != null && output.getCatalogUri() != null) {
+      catalogClient.validate(
+          output.getOutputCatalog(),
+          output.getCatalogUri(),
+          new ArrayList<>());
+    }
+
     try {
       final var req = new FailAttemptRequest()
           .attemptNumber(input.getAttemptNumber())

diff --git a/airbyte-workers/src/main/java/io/airbyte/workers/temporal/sync/ReplicationActivityImpl.java b/airbyte-workers/src/main/java/io/airbyte/workers/temporal/sync/ReplicationActivityImpl.java
@@ -31,9 +31,7 @@
 import io.airbyte.featureflag.Connection;
 import io.airbyte.featureflag.FeatureFlagClient;
 import io.airbyte.featureflag.NullOutputCatalogOnSyncOutput;
-import io.airbyte.featureflag.NullOutputStateOnSyncOutput;
 import io.airbyte.featureflag.WriteOutputCatalogToObjectStorage;
-import io.airbyte.featureflag.WriteOutputStateToObjectStorage;
 import io.airbyte.metrics.lib.ApmTraceUtils;
 import io.airbyte.metrics.lib.MetricAttribute;
 import io.airbyte.metrics.lib.MetricClient;
@@ -225,17 +223,6 @@ public StandardSyncOutput replicateV2(final ReplicationActivityInput replication
           BackfillHelper.markBackfilledStreams(streamsToBackfill, standardSyncOutput);
           LOGGER.info("sync summary after backfill: {}", standardSyncOutput);
 
-          if (featureFlagClient.boolVariation(WriteOutputStateToObjectStorage.INSTANCE, new Connection(connectionId))) {
-            final var uri = stateStorageClient.persist(
-                attemptOutput.getState(),
-                connectionId,
-                Long.parseLong(jobId),
-                attemptNumber.intValue(),
-                metricAttributes);
-
-            standardSyncOutput.setStateUri(uri);
-          }
-
           if (featureFlagClient.boolVariation(WriteOutputCatalogToObjectStorage.INSTANCE, new Connection(connectionId))) {
             final var uri = catalogStorageClient.persist(
                 attemptOutput.getOutputCatalog(),
@@ -273,10 +260,6 @@ private StandardSyncOutput reduceReplicationOutput(final ReplicationOutput outpu
     syncSummary.setPerformanceMetrics(output.getReplicationAttemptSummary().getPerformanceMetrics());
     syncSummary.setStreamCount((long) output.getOutputCatalog().getStreams().size());
 
-    if (!featureFlagClient.boolVariation(NullOutputStateOnSyncOutput.INSTANCE, new Connection(connectionId))) {
-      standardSyncOutput.setState(output.getState());
-    }
-
     if (!featureFlagClient.boolVariation(NullOutputCatalogOnSyncOutput.INSTANCE, new Connection(connectionId))) {
       standardSyncOutput.setOutputCatalog(output.getOutputCatalog());
     }