From f2560a850f812f315d1a0f9c80419848b80016fd Mon Sep 17 00:00:00 2001
From: Yue Li <61070669+theyueli@users.noreply.github.com>
Date: Tue, 28 May 2024 18:23:14 -0700
Subject: [PATCH] CDK: make postgres temp_file_limit error a transient error
 (#38734)

Fixes https://github.com/airbytehq/airbyte/issues/27090

Postgres throws a temp_file_limit error is translated to a system error for Airbyte. I believe this is true for both Postgres source and destination connectors.

This change makes this error transient so that it won't trigger sentry. It also fixes a bug that should use a lower-case version of a transient error message.
---
 .../util/ConnectorExceptionUtil.kt            | 29 ++++++++++++++++---
 1 file changed, 25 insertions(+), 4 deletions(-)

diff --git a/airbyte-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/integrations/util/ConnectorExceptionUtil.kt b/airbyte-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/integrations/util/ConnectorExceptionUtil.kt
index dba7ae8c4eef1..8009650017d27 100644
--- a/airbyte-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/integrations/util/ConnectorExceptionUtil.kt
+++ b/airbyte-cdk/java/airbyte-cdk/core/src/main/kotlin/io/airbyte/cdk/integrations/util/ConnectorExceptionUtil.kt
@@ -130,6 +130,16 @@ object ConnectorExceptionUtil {
         return eithers.map { obj: Either<out T, Result> -> obj.right!! }
     }
 
+    private val TRANSIENT_SQL_EXCEPTION_MESSAGE: Array<String> =
+        arrayOf(
+            "an i/o error occurred while sending to the backend",
+            "temporary file size exceeds temp_file_limit"
+        )
+    private val TRANSIENT_EOF_EXCEPTION_MESSAGE: Array<String> =
+        arrayOf("connection was unexpectedly lost")
+    private val RECOVERY_CONNECTION_EXCEPTION_MESSAGE: Array<String> =
+        arrayOf("due to conflict with recovery")
+
     private fun isTransientErrorException(e: Throwable?): Boolean {
         return e is TransientErrorException
     }
@@ -142,19 +152,30 @@ object ConnectorExceptionUtil {
         return e is ConnectionErrorException
     }
 
+    private fun containsOneOfTheErrorMessages(
+        e: Throwable?,
+        errorMessages: Array<String>
+    ): Boolean {
+        val msg = e?.message!!.lowercase()
+        for (errorMessage in errorMessages) {
+            if (msg.contains(errorMessage)) return true
+        }
+        return false
+    }
+
     private fun isTransientEOFException(e: Throwable?): Boolean {
         return (e is EOFException) &&
-            e.message!!.lowercase().contains("connection was unexpectedly lost")
+            containsOneOfTheErrorMessages(e, TRANSIENT_EOF_EXCEPTION_MESSAGE)
     }
 
     private fun isTransientSQLException(e: Throwable?): Boolean {
         return (e is SQLException) &&
-            e.message!!.lowercase().contains("An I/O error occurred while sending to the backend")
+            containsOneOfTheErrorMessages(e, TRANSIENT_SQL_EXCEPTION_MESSAGE)
     }
 
     private fun isRecoveryConnectionException(e: Throwable?): Boolean {
-        return e is SQLException &&
-            e.message!!.lowercase().contains("due to conflict with recovery")
+        return (e is SQLException) &&
+            containsOneOfTheErrorMessages(e, RECOVERY_CONNECTION_EXCEPTION_MESSAGE)
     }
 
     private fun isUnknownColumnInFieldListException(e: Throwable?): Boolean {