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

Author: akashkulk

airbyte-cdk/java/airbyte-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(

airbyte-integrations/connectors/source-postgres/gradle.properties

@@ -1,3 +1,3 @@
 testExecutionConcurrency=-1
 
-JunitMethodExecutionTimeout=2 m
+JunitMethodExecutionTimeout=5 m

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

airbyte-integrations/connectors/source-postgres/src/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());

airbyte-integrations/connectors/source-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");