JDBC Destinations: improve error message for conflicting streams

airbyte-config/init/src/main/resources/seed/destination_definitions.yaml

@@ -441,4 +441,4 @@
   dockerRepository: airbyte/destination-weaviate
   dockerImageTag: 0.1.0
   documentationUrl: https://docs.airbyte.com/integrations/destinations/weaviate
-  releaseStage: alpha
+  releaseStage: alpha

airbyte-integrations/connectors/destination-jdbc/src/main/java/io/airbyte/integrations/destination/staging/StagingConsumerFactory.java

@@ -4,8 +4,13 @@
 
 package io.airbyte.integrations.destination.staging;
 
+import static java.util.stream.Collectors.joining;
+import static java.util.stream.Collectors.toList;
+
 import com.fasterxml.jackson.databind.JsonNode;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
+import io.airbyte.commons.exceptions.ConfigErrorException;
 import io.airbyte.commons.functional.CheckedBiConsumer;
 import io.airbyte.commons.functional.CheckedBiFunction;
 import io.airbyte.commons.json.Jsons;
@@ -25,12 +30,14 @@
 import io.airbyte.protocol.models.v0.ConfiguredAirbyteStream;
 import io.airbyte.protocol.models.v0.DestinationSyncMode;
 import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.UUID;
 import java.util.function.Consumer;
 import java.util.function.Function;
-import java.util.stream.Collectors;
 import org.apache.commons.io.FileUtils;
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
@@ -89,7 +96,7 @@ private static List<WriteConfig> createWriteConfigs(final NamingConventionTransf
                                                       final JsonNode config,
                                                       final ConfiguredAirbyteCatalog catalog) {
 
-    return catalog.getStreams().stream().map(toWriteConfig(namingResolver, config)).collect(Collectors.toList());
+    return catalog.getStreams().stream().map(toWriteConfig(namingResolver, config)).collect(toList());
   }
 
   private static Function<ConfiguredAirbyteStream, WriteConfig> toWriteConfig(final NamingConventionTransformer namingResolver,
@@ -160,15 +167,32 @@ private static AirbyteStreamNameNamespacePair toNameNamespacePair(final WriteCon
    * @param catalog collection of configured streams (e.g. API endpoints or database tables)
    * @return
    */
-  private CheckedBiConsumer<AirbyteStreamNameNamespacePair, SerializableBuffer, Exception> flushBufferFunction(
+  @VisibleForTesting
+  CheckedBiConsumer<AirbyteStreamNameNamespacePair, SerializableBuffer, Exception> flushBufferFunction(
                                                                                                                final JdbcDatabase database,
                                                                                                                final StagingOperations stagingOperations,
                                                                                                                final List<WriteConfig> writeConfigs,
                                                                                                                final ConfiguredAirbyteCatalog catalog) {
-    final Map<AirbyteStreamNameNamespacePair, WriteConfig> pairToWriteConfig =
-        writeConfigs.stream()
-            .collect(Collectors.toUnmodifiableMap(
-                StagingConsumerFactory::toNameNamespacePair, Function.identity()));
+    final Set<WriteConfig> conflictingStreams = new HashSet<>();
+    final Map<AirbyteStreamNameNamespacePair, WriteConfig> pairToWriteConfig = new HashMap<>();
+    for (final WriteConfig config : writeConfigs) {
+      final AirbyteStreamNameNamespacePair streamIdentifier = toNameNamespacePair(config);
+      if (pairToWriteConfig.containsKey(streamIdentifier)) {
+        conflictingStreams.add(config);
+        final WriteConfig existingConfig = pairToWriteConfig.get(streamIdentifier);
+        // The first conflicting stream won't have any problems, so we need to explicitly add it here.
+        conflictingStreams.add(existingConfig);
+      } else {
+        pairToWriteConfig.put(streamIdentifier, config);
+      }
+    }
+    if (!conflictingStreams.isEmpty()) {
+      final String message = String.format(
+          "You are trying to write multiple streams to the same table. Consider switching to a custom namespace format using ${SOURCE_NAMESPACE}, or moving one of them into a separate connection with a different stream prefix. Affected streams: %s",
+          conflictingStreams.stream().map(config -> config.getNamespace() + "." + config.getNamespace()).collect(joining(", "))
+      );
+      throw new ConfigErrorException(message);
+    }
 
     return (pair, writer) -> {
       LOGGER.info("Flushing buffer for stream {} ({}) to staging", pair.getName(), FileUtils.byteCountToDisplaySize(writer.getByteCount()));

airbyte-integrations/connectors/destination-jdbc/src/test/java/io/airbyte/integrations/destination/staging/StagingConsumerFactoryTest.java

@@ -0,0 +1,34 @@
+package io.airbyte.integrations.destination.staging;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+import io.airbyte.commons.exceptions.ConfigErrorException;
+import io.airbyte.integrations.destination.jdbc.WriteConfig;
+import java.util.List;
+import org.junit.jupiter.api.Test;
+
+class StagingConsumerFactoryTest {
+
+  @Test()
+  void detectConflictingStreams() {
+    final StagingConsumerFactory f = new StagingConsumerFactory();
+
+    final ConfigErrorException configErrorException = assertThrows(
+        ConfigErrorException.class,
+        () -> f.flushBufferFunction(
+            null,
+            null,
+            List.of(
+                new WriteConfig("source_schema1", "example_stream", "destination_default_schema", null, null, null),
+                new WriteConfig("source_schema2", "example_stream", "destination_default_schema", null, null, null)
+            ),
+            null
+        ));
+
+    assertEquals(
+        "You are trying to write multiple streams to the same table. Consider switching to a custom namespace format using ${SOURCE_NAMESPACE}, or moving one of them into a separate connection with a different stream prefix. Affected streams: source_schema1.example_stream, source_schema2.example_stream",
+        configErrorException.getMessage()
+    );
+  }
+
+}