[Source-postgres] : Better error messages on switching between sync m… (

#37509)
airbytehq · Apr 24, 2024 · d790b7d · d790b7d
1 parent b5fef4e
commit d790b7d
Show file tree

Hide file tree

Showing 6 changed files with 242 additions and 234 deletions.
diff --git a/...e-cdk/core/src/main/kotlin/io/airbyte/cdk/integrations/base/AirbyteTraceMessageUtility.kt b/...e-cdk/core/src/main/kotlin/io/airbyte/cdk/integrations/base/AirbyteTraceMessageUtility.kt
@@ -19,6 +19,11 @@ object AirbyteTraceMessageUtility {
         emitErrorTrace(e, displayMessage, AirbyteErrorTraceMessage.FailureType.CONFIG_ERROR)
     }
 
+    @JvmStatic
+    fun emitTransientErrorTrace(e: Throwable, displayMessage: String?) {
+        emitErrorTrace(e, displayMessage, AirbyteErrorTraceMessage.FailureType.SYSTEM_ERROR)
+    }
+
     fun emitCustomErrorTrace(displayMessage: String?, internalMessage: String?) {
         emitMessage(
             makeAirbyteMessageFromTraceMessage(

diff --git a/airbyte-integrations/connectors/source-postgres/gradle.properties b/airbyte-integrations/connectors/source-postgres/gradle.properties
@@ -1,3 +1,3 @@
 testExecutionConcurrency=-1
 
-JunitMethodExecutionTimeout=2 m
+JunitMethodExecutionTimeout=5 m
diff --git a/airbyte-integrations/connectors/source-postgres/metadata.yaml b/airbyte-integrations/connectors/source-postgres/metadata.yaml
@@ -9,7 +9,7 @@ data:
   connectorSubtype: database
   connectorType: source
   definitionId: decd338e-5647-4c0b-adf4-da0e75f5a750
-  dockerImageTag: 3.3.27
+  dockerImageTag: 3.3.28
   dockerRepository: airbyte/source-postgres
   documentationUrl: https://docs.airbyte.com/integrations/sources/postgres
   githubIssueLabel: source-postgres

diff --git a/.../main/java/io/airbyte/integrations/source/postgres/cursor_based/CursorBasedCtidUtils.java b/.../main/java/io/airbyte/integrations/source/postgres/cursor_based/CursorBasedCtidUtils.java
@@ -10,6 +10,7 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 import io.airbyte.cdk.integrations.source.relationaldb.state.StateManager;
+import io.airbyte.commons.exceptions.ConfigErrorException;
 import io.airbyte.integrations.source.postgres.ctid.CtidUtils.CtidStreams;
 import io.airbyte.integrations.source.postgres.ctid.CtidUtils.StreamsCategorised;
 import io.airbyte.integrations.source.postgres.internal.models.InternalModels.StateType;
@@ -67,7 +68,7 @@ public static StreamsCategorised<CursorBasedStreams> categoriseStreams(final Sta
             cursorBasedSyncStreamPairs.add(pair);
             statesFromCursorBasedSync.add(stateMessage);
           } else {
-            throw new RuntimeException("Unknown state type: " + streamState.get(STATE_TYPE_KEY).asText());
+            throw new ConfigErrorException("You've changed replication modes - please reset the streams in this connector");
           }
         } else {
           LOGGER.info("State type not present, syncing stream {} via cursor", streamDescriptor.getName());

diff --git a/...ce-postgres/src/main/java/io/airbyte/integrations/source/postgres/xmin/XminCtidUtils.java b/...ce-postgres/src/main/java/io/airbyte/integrations/source/postgres/xmin/XminCtidUtils.java
@@ -10,6 +10,7 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 import io.airbyte.cdk.integrations.source.relationaldb.state.StateManager;
+import io.airbyte.commons.exceptions.ConfigErrorException;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.integrations.source.postgres.ctid.CtidUtils.CtidStreams;
 import io.airbyte.integrations.source.postgres.ctid.CtidUtils.StreamsCategorised;
@@ -72,7 +73,7 @@ public static StreamsCategorised<XminStreams> categoriseStreams(final StateManag
               statesFromXminSync.add(stateMessage);
             }
           } else {
-            throw new RuntimeException("Unknown state type: " + streamState.get(STATE_TYPE_KEY).asText());
+            throw new ConfigErrorException("You've changed replication modes - please reset the streams in this connector");
           }
         } else {
           throw new RuntimeException("State type not present");