Properly set catalog and state fields. Handle null state in validate.…

… (#12086)
airbytehq · Apr 11, 2024 · e994f43 · e994f43
1 parent 4e5508b
commit e994f43
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 3 deletions.
diff --git a/...mmons-worker/src/main/kotlin/io/airbyte/workers/storage/activities/OutputStorageClient.kt b/...mmons-worker/src/main/kotlin/io/airbyte/workers/storage/activities/OutputStorageClient.kt
@@ -61,12 +61,14 @@ class OutputStorageClient<T : Any>
      * the expected. Emits a metric whether it's a match.
      */
     fun validate(
-      expected: T,
+      expected: T?,
       connectionId: UUID,
       jobId: Long,
       attemptNumber: Int,
       attrs: List<MetricAttribute>,
     ) {
+      if (expected == null) return
+
       val uri = ActivityPayloadURI.v1(connectionId, jobId, attemptNumber, payloadName)
 
       storageClient.validateOutput(

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
@@ -270,11 +270,11 @@ private StandardSyncOutput reduceReplicationOutput(final ReplicationOutput outpu
     syncSummary.setStreamCount((long) output.getOutputCatalog().getStreams().size());
 
     if (!featureFlagClient.boolVariation(NullOutputStateOnSyncOutput.INSTANCE, new Connection(connectionId))) {
-      standardSyncOutput.setState(standardSyncOutput.getState());
+      standardSyncOutput.setState(output.getState());
     }
 
     if (!featureFlagClient.boolVariation(NullOutputCatalogOnSyncOutput.INSTANCE, new Connection(connectionId))) {
-      standardSyncOutput.setOutputCatalog(standardSyncOutput.getOutputCatalog());
+      standardSyncOutput.setOutputCatalog(output.getOutputCatalog());
     }
 
     standardSyncOutput.setStandardSyncSummary(syncSummary);