Migrating InvalidCursorException -> ConfigErrorException (#18995)

* Migrate InvalidCursorInfo -> ConfigErrorException

Author: akashkulk

airbyte-integrations/connectors/source-postgres/src/test/java/io/airbyte/integrations/source/postgres/PostgresSourceTest.java

@@ -18,6 +18,7 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
+import io.airbyte.commons.exceptions.ConfigErrorException;
 import io.airbyte.commons.io.IOs;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.commons.string.Strings;
@@ -26,7 +27,6 @@
 import io.airbyte.db.factory.DSLContextFactory;
 import io.airbyte.db.factory.DatabaseDriver;
 import io.airbyte.db.jdbc.JdbcUtils;
-import io.airbyte.integrations.source.relationaldb.InvalidCursorException;
 import io.airbyte.protocol.models.AirbyteCatalog;
 import io.airbyte.protocol.models.AirbyteMessage;
 import io.airbyte.protocol.models.AirbyteStream;
@@ -526,7 +526,7 @@ public void tableWithInvalidCursorShouldThrowException() throws Exception {
             new ConfiguredAirbyteCatalog().withStreams(Collections.singletonList(tableWithInvalidCursorType));
 
         final Throwable throwable = catchThrowable(() -> MoreIterators.toSet(new PostgresSource().read(config, configuredAirbyteCatalog, null)));
-        assertThat(throwable).isInstanceOf(InvalidCursorException.class)
+        assertThat(throwable).isInstanceOf(ConfigErrorException.class)
             .hasMessageContaining(
                 "The following tables have invalid columns selected as cursor, please select a column with a well-defined ordering as a cursor. {tableName='public.test_table', cursorColumnName='id', cursorSqlType=OTHER}");
       } finally {

airbyte-integrations/connectors/source-relational-db/src/main/java/io/airbyte/integrations/source/relationaldb/AbstractDbSource.java

@@ -9,6 +9,7 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Lists;
+import io.airbyte.commons.exceptions.ConfigErrorException;
 import io.airbyte.commons.exceptions.ConnectionErrorException;
 import io.airbyte.commons.features.EnvVariableFeatureFlags;
 import io.airbyte.commons.features.FeatureFlags;
@@ -26,7 +27,7 @@
 import io.airbyte.integrations.base.AirbyteStreamNameNamespacePair;
 import io.airbyte.integrations.base.AirbyteTraceMessageUtility;
 import io.airbyte.integrations.base.Source;
-import io.airbyte.integrations.source.relationaldb.InvalidCursorException.InvalidCursorInfo;
+import io.airbyte.integrations.source.relationaldb.InvalidCursorInfoUtil.InvalidCursorInfo;
 import io.airbyte.integrations.source.relationaldb.models.DbState;
 import io.airbyte.integrations.source.relationaldb.state.StateManager;
 import io.airbyte.integrations.source.relationaldb.state.StateManagerFactory;
@@ -142,56 +143,42 @@ public AirbyteCatalog discover(final JsonNode config) throws Exception {
    */
   @Override
   public AutoCloseableIterator<AirbyteMessage> read(final JsonNode config,
-      final ConfiguredAirbyteCatalog catalog,
-      final JsonNode state)
-      throws Exception {
-    try {
-      final StateManager stateManager =
-          StateManagerFactory.createStateManager(getSupportedStateType(config),
-              deserializeInitialState(state, config), catalog);
-      final Instant emittedAt = Instant.now();
-
-      final Database database = createDatabaseInternal(config);
-
-      final Map<String, TableInfo<CommonField<DataType>>> fullyQualifiedTableNameToInfo =
-          discoverWithoutSystemTables(database)
-              .stream()
-              .collect(Collectors.toMap(t -> String.format("%s.%s", t.getNameSpace(), t.getName()),
-                  Function
-                      .identity()));
-
-      validateCursorFieldForIncrementalTables(fullyQualifiedTableNameToInfo, catalog);
-
-      final List<AutoCloseableIterator<AirbyteMessage>> incrementalIterators =
-          getIncrementalIterators(database, catalog, fullyQualifiedTableNameToInfo, stateManager,
-              emittedAt);
-      final List<AutoCloseableIterator<AirbyteMessage>> fullRefreshIterators =
-          getFullRefreshIterators(database, catalog, fullyQualifiedTableNameToInfo, stateManager,
-              emittedAt);
-      final List<AutoCloseableIterator<AirbyteMessage>> iteratorList = Stream
-          .of(incrementalIterators, fullRefreshIterators)
-          .flatMap(Collection::stream)
-          .collect(Collectors.toList());
-
-      return AutoCloseableIterators
-          .appendOnClose(AutoCloseableIterators.concatWithEagerClose(iteratorList), () -> {
-            LOGGER.info("Closing database connection pool.");
-            Exceptions.toRuntime(this::close);
-            LOGGER.info("Closed database connection pool.");
-          });
-    } catch (final Exception exception) {
-      if (isConfigError(exception)) {
-        AirbyteTraceMessageUtility.emitConfigErrorTrace(exception, exception.getMessage());
-      }
-      throw exception;
-    }
-  }
+    final ConfiguredAirbyteCatalog catalog,
+    final JsonNode state)
+    throws Exception {
+    final StateManager stateManager =
+        StateManagerFactory.createStateManager(getSupportedStateType(config),
+            deserializeInitialState(state, config), catalog);
+    final Instant emittedAt = Instant.now();
+
+    final Database database = createDatabaseInternal(config);
+
+    final Map<String, TableInfo<CommonField<DataType>>> fullyQualifiedTableNameToInfo =
+        discoverWithoutSystemTables(database)
+            .stream()
+            .collect(Collectors.toMap(t -> String.format("%s.%s", t.getNameSpace(), t.getName()),
+                Function
+                    .identity()));
+
+    validateCursorFieldForIncrementalTables(fullyQualifiedTableNameToInfo, catalog);
+
+    final List<AutoCloseableIterator<AirbyteMessage>> incrementalIterators =
+        getIncrementalIterators(database, catalog, fullyQualifiedTableNameToInfo, stateManager,
+            emittedAt);
+    final List<AutoCloseableIterator<AirbyteMessage>> fullRefreshIterators =
+        getFullRefreshIterators(database, catalog, fullyQualifiedTableNameToInfo, stateManager,
+            emittedAt);
+    final List<AutoCloseableIterator<AirbyteMessage>> iteratorList = Stream
+        .of(incrementalIterators, fullRefreshIterators)
+        .flatMap(Collection::stream)
+        .collect(Collectors.toList());
 
-  private boolean isConfigError(final Exception exception) {
-    // For now, enhanced error details should only be shown for InvalidCursorException. In the future,
-    // enhanced error messages will exist for
-    // additional error types.
-    return exception instanceof InvalidCursorException;
+    return AutoCloseableIterators
+        .appendOnClose(AutoCloseableIterators.concatWithEagerClose(iteratorList), () -> {
+          LOGGER.info("Closing database connection pool.");
+          Exceptions.toRuntime(this::close);
+          LOGGER.info("Closed database connection pool.");
+        });
   }
 
   private void validateCursorFieldForIncrementalTables(
@@ -230,7 +217,8 @@ private void validateCursorFieldForIncrementalTables(
     }
 
     if (!tablesWithInvalidCursor.isEmpty()) {
-      throw new InvalidCursorException(tablesWithInvalidCursor);
+      throw new ConfigErrorException(
+          InvalidCursorInfoUtil.getInvalidCursorConfigMessage(tablesWithInvalidCursor)) ;
     }
   }
 

airbyte-integrations/connectors/source-relational-db/src/main/java/io/airbyte/integrations/source/relationaldb/InvalidCursorException.java renamed to airbyte-integrations/connectors/source-relational-db/src/main/java/io/airbyte/integrations/source/relationaldb/InvalidCursorInfoUtil.java

@@ -7,12 +7,12 @@
 import java.util.List;
 import java.util.stream.Collectors;
 
-public class InvalidCursorException extends RuntimeException {
+public class InvalidCursorInfoUtil {
 
-  public InvalidCursorException(final List<InvalidCursorInfo> tablesWithInvalidCursor) {
-    super("The following tables have invalid columns selected as cursor, please select a column with a well-defined ordering as a cursor. "
+  public static String getInvalidCursorConfigMessage(final List<InvalidCursorInfo> tablesWithInvalidCursor) {
+    return "The following tables have invalid columns selected as cursor, please select a column with a well-defined ordering as a cursor. "
         + tablesWithInvalidCursor.stream().map(InvalidCursorInfo::toString)
-            .collect(Collectors.joining(",")));
+            .collect(Collectors.joining(","));
   }
 
   public record InvalidCursorInfo(String tableName, String cursorColumnName, String cursorSqlType) {