Destinations: throw error on empty catalog (#39552)

Author: edgao

airbyte-cdk/java/airbyte-cdk/README.md

@@ -174,6 +174,7 @@ corresponds to that version.
 
 | Version | Date       | Pull Request                                               | Subject                                                                                                                                                        |
 |:--------| :--------- | :--------------------------------------------------------- | :------------------------------------------------------------------------------------------------------------------------------------------------------------- |
+| 0.40.2  | 2024-06-18 | [\#39552](https://github.com/airbytehq/airbyte/pull/39552) | Destinations: Throw error if the ConfiguredCatalog has no streams                                                                                              |
 | 0.40.1  | 2024-06-14 | [\#39349](https://github.com/airbytehq/airbyte/pull/39349) | Source stats for full refresh streams                                                                                                                          |
 | 0.40.0  | 2024-06-17 | [\#38622](https://github.com/airbytehq/airbyte/pull/38622) | Destinations: Implement refreshes logic in AbstractStreamOperation                                                                                             |
 | 0.39.0  | 2024-06-17 | [\#38067](https://github.com/airbytehq/airbyte/pull/38067) | Destinations: Breaking changes for refreshes (fail on INCOMPLETE stream status; ignore OVERWRITE sync mode)                                                    |

airbyte-cdk/java/airbyte-cdk/core/src/main/resources/version.properties

@@ -1 +1 @@
-version=0.40.1
+version=0.40.2

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

@@ -25,10 +25,16 @@ constructor(
     private val defaultNamespace: String,
     private val rawNamespace: String = JavaBaseConstants.DEFAULT_AIRBYTE_INTERNAL_NAMESPACE,
 ) {
-    fun parseCatalog(orginalCatalog: ConfiguredAirbyteCatalog): ParsedCatalog {
+    fun parseCatalog(originalCatalog: ConfiguredAirbyteCatalog): ParsedCatalog {
+        if (originalCatalog.streams.isEmpty()) {
+            throw ConfigErrorException(
+                "The catalog contained no streams. This likely indicates a platform/configuration error."
+            )
+        }
+
         // Don't mutate the original catalog, just operate on a copy of it
         // This is... probably the easiest way we have to deep clone a protocol model object?
-        val catalog = Jsons.clone(orginalCatalog)
+        val catalog = Jsons.clone(originalCatalog)
         catalog.streams.onEach {
             // Overwrite null namespaces
             if (it.stream.namespace.isNullOrEmpty()) {

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

@@ -4,6 +4,7 @@
 package io.airbyte.integrations.base.destination.typing_deduping
 
 import com.fasterxml.jackson.databind.JsonNode
+import io.airbyte.commons.exceptions.ConfigErrorException
 import io.airbyte.commons.json.Jsons
 import io.airbyte.protocol.models.v0.AirbyteStream
 import io.airbyte.protocol.models.v0.ConfiguredAirbyteCatalog
@@ -12,6 +13,7 @@ import io.airbyte.protocol.models.v0.DestinationSyncMode
 import io.airbyte.protocol.models.v0.SyncMode
 import org.junit.jupiter.api.Assertions
 import org.junit.jupiter.api.Assertions.assertAll
+import org.junit.jupiter.api.Assertions.assertThrows
 import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.Test
 import org.mockito.Mockito
@@ -48,6 +50,13 @@ internal class CatalogParserTest {
         parser = CatalogParser(sqlGenerator, "default_namespace")
     }
 
+    @Test
+    fun throwOnEmptyCatalog() {
+        assertThrows(ConfigErrorException::class.java) {
+            parser.parseCatalog(ConfiguredAirbyteCatalog().withStreams(emptyList()))
+        }
+    }
+
     /**
      * Both these streams will write to the same final table name ("foofoo"). Verify that they don't
      * actually use the same tablename.