throw exception on bad generation id

Author: edgao

airbyte-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/integrations/destination/async/AsyncStreamConsumer.kt

@@ -164,11 +164,10 @@ constructor(
 
         val streamSyncSummaries =
             streamNames.associate { streamDescriptor ->
-                // If we didn't receive a stream status message, assume success.
-                // Platform won't send us any stream status messages yet (since we're not declaring
-                // supportsRefresh in metadata), so we will always hit this case.
+                // If we didn't receive a stream status message, assume failure.
+                // This is possible if e.g. the orchestrator crashes before sending us the message.
                 val terminalStatusFromSource =
-                    terminalStatusesFromSource[streamDescriptor] ?: AirbyteStreamStatus.COMPLETE
+                    terminalStatusesFromSource[streamDescriptor] ?: AirbyteStreamStatus.INCOMPLETE
                 StreamDescriptorUtils.withDefaultNamespace(
                     streamDescriptor,
                     bufferManager.defaultNamespace,

airbyte-cdk/java/airbyte-cdk/core/src/test/kotlin/io/airbyte/cdk/integrations/destination/async/AsyncStreamConsumerTest.kt

@@ -553,8 +553,7 @@ class AsyncStreamConsumerTest {
             ArgumentCaptor.captor()
         Mockito.verify(onClose).accept(any(), capture(captor))
         assertEquals(
-            // All streams have a COMPLETE status.
-            // TODO: change this to INCOMPLETE after we switch the default behavior.
+            // All streams have an INCOMPLETE status.
             mapOf(
                 StreamDescriptor().withNamespace(SCHEMA_NAME).withName(STREAM_NAME) to
                     StreamSyncSummary(

airbyte-cdk/java/airbyte-cdk/typing-deduping/src/main/kotlin/io/airbyte/integrations/base/destination/typing_deduping/CatalogParser.kt

@@ -6,9 +6,11 @@ package io.airbyte.integrations.base.destination.typing_deduping
 import com.google.common.annotations.VisibleForTesting
 import io.airbyte.cdk.integrations.base.AirbyteExceptionHandler.Companion.addStringForDeinterpolation
 import io.airbyte.cdk.integrations.base.JavaBaseConstants
+import io.airbyte.commons.exceptions.ConfigErrorException
 import io.airbyte.commons.json.Jsons
 import io.airbyte.protocol.models.v0.ConfiguredAirbyteCatalog
 import io.airbyte.protocol.models.v0.ConfiguredAirbyteStream
+import io.airbyte.protocol.models.v0.DestinationSyncMode
 import java.util.Optional
 import java.util.function.Consumer
 import org.apache.commons.codec.digest.DigestUtils
@@ -31,6 +33,12 @@ constructor(
             if (it.stream.namespace.isNullOrEmpty()) {
                 it.stream.namespace = defaultNamespace
             }
+            // The refreshes project is the beginning of the end for OVERWRITE syncs.
+            // The sync mode still exists, but we are fully dependent on min_generation to trigger
+            // overwrite logic.
+            if (it.destinationSyncMode == DestinationSyncMode.OVERWRITE) {
+                it.destinationSyncMode = DestinationSyncMode.APPEND
+            }
         }
 
         // this code is bad and I feel bad
@@ -125,9 +133,10 @@ constructor(
     @VisibleForTesting
     fun toStreamConfig(stream: ConfiguredAirbyteStream): StreamConfig {
         if (stream.generationId == null) {
-            stream.generationId = 0
-            stream.minimumGenerationId = 0
-            stream.syncId = 0
+            // TODO set platform version
+            throw ConfigErrorException(
+                "You must upgrade your platform version to use this connector version. Either downgrade your connector or upgrade platform to X.Y.Z"
+            )
         }
         if (
             stream.minimumGenerationId != 0.toLong() &&