fix(destination-bigquery): handle InterruptedException during sync cancellation

airbyte-integrations/connectors/destination-bigquery/metadata.yaml

@@ -6,7 +6,7 @@ data:
   connectorSubtype: database
   connectorType: destination
   definitionId: 22f6c74f-5699-40ff-833c-4a879ea40133
-  dockerImageTag: 3.0.19
+  dockerImageTag: 3.0.20
   dockerRepository: airbyte/destination-bigquery
   documentationUrl: https://docs.airbyte.com/integrations/destinations/bigquery
   githubIssueLabel: destination-bigquery

airbyte-integrations/connectors/destination-bigquery/src/main/kotlin/io/airbyte/integrations/destination/bigquery/BigQueryUtils.kt

@@ -10,6 +10,7 @@ import com.google.cloud.bigquery.*
 import com.google.common.collect.ImmutableList
 import com.google.common.collect.ImmutableMap
 import io.airbyte.cdk.ConfigErrorException
+import io.airbyte.cdk.TransientErrorException
 import io.airbyte.cdk.load.message.Meta
 import java.util.*
 import java.util.stream.Collectors
@@ -176,6 +177,27 @@ object BigQueryUtils {
             Optional.ofNullable(System.getenv("WORKER_CONNECTOR_IMAGE"))
                 .map { name: String -> name.replace("airbyte/", "").replace(":", "/") }
                 .orElse("destination-bigquery")
+
+    /**
+     * Executes a BigQuery operation, converting [BigQueryException] caused by
+     * [InterruptedException] into a [TransientErrorException]. This prevents raw Java exception
+     * class names from surfacing to users when the platform cancels a sync.
+     */
+    @JvmStatic
+    fun <T> executeBigQueryOperation(operation: () -> T): T {
+        try {
+            return operation()
+        } catch (e: BigQueryException) {
+            if (e.cause is InterruptedException) {
+                Thread.currentThread().interrupt()
+                throw TransientErrorException(
+                    "BigQuery API call interrupted.",
+                    e,
+                )
+            }
+            throw e
+        }
+    }
 }
 
 fun TableId.toPrettyString() = "${this.dataset}.${this.table}"

airbyte-integrations/connectors/destination-bigquery/src/main/kotlin/io/airbyte/integrations/destination/bigquery/write/typing_deduping/direct_load_tables/BigqueryDirectLoadDatabaseInitialStatusGatherer.kt

@@ -13,6 +13,7 @@ import io.airbyte.cdk.load.orchestration.db.TempTableNameGenerator
 import io.airbyte.cdk.load.orchestration.db.direct_load_table.DirectLoadInitialStatus
 import io.airbyte.cdk.load.orchestration.db.direct_load_table.DirectLoadTableStatus
 import io.airbyte.cdk.load.orchestration.db.legacy_typing_deduping.TableCatalog
+import io.airbyte.integrations.destination.bigquery.BigQueryUtils
 import io.airbyte.integrations.destination.bigquery.write.typing_deduping.toTableId
 import java.math.BigInteger
 import java.util.concurrent.ConcurrentHashMap
@@ -44,7 +45,8 @@ class BigqueryDirectLoadDatabaseInitialStatusGatherer(
     }
 
     private fun getTableStatus(tableName: TableName): DirectLoadTableStatus? {
-        val table = bigquery.getTable(tableName.toTableId())
+        val table =
+            BigQueryUtils.executeBigQueryOperation { bigquery.getTable(tableName.toTableId()) }
         return table?.let { DirectLoadTableStatus(isEmpty = table.numRows == BigInteger.ZERO) }
     }
 }

airbyte-integrations/connectors/destination-bigquery/src/main/kotlin/io/airbyte/integrations/destination/bigquery/write/typing_deduping/legacy_raw_tables/BigqueryTypingDedupingDatabaseInitialStatusGatherer.kt

@@ -15,32 +15,37 @@ import io.airbyte.cdk.load.orchestration.db.legacy_typing_deduping.FinalTableIni
 import io.airbyte.cdk.load.orchestration.db.legacy_typing_deduping.RawTableInitialStatus
 import io.airbyte.cdk.load.orchestration.db.legacy_typing_deduping.TableCatalog
 import io.airbyte.cdk.load.orchestration.db.legacy_typing_deduping.TypingDedupingDatabaseInitialStatus
+import io.airbyte.integrations.destination.bigquery.BigQueryUtils
 
 class BigqueryTypingDedupingDatabaseInitialStatusGatherer(private val bq: BigQuery) :
     DatabaseInitialStatusGatherer<TypingDedupingDatabaseInitialStatus> {
     private fun getInitialRawTableState(
         rawTableName: TableName,
         suffix: String
     ): RawTableInitialStatus? {
-        bq.getTable(TableId.of(rawTableName.namespace, rawTableName.name + suffix))
+        BigQueryUtils.executeBigQueryOperation {
+            bq.getTable(TableId.of(rawTableName.namespace, rawTableName.name + suffix))
+        }
         // Table doesn't exist. There are no unprocessed records, and no timestamp.
         ?: return null
 
         val rawTableIdQuoted = """`${rawTableName.namespace}`.`${rawTableName.name}$suffix`"""
         val unloadedRecordTimestamp =
-            bq.query(
-                    QueryJobConfiguration.of(
-                        """
+            BigQueryUtils.executeBigQueryOperation {
+                bq.query(
+                        QueryJobConfiguration.of(
+                            """
                             SELECT TIMESTAMP_SUB(MIN(_airbyte_extracted_at), INTERVAL 1 MICROSECOND)
                             FROM $rawTableIdQuoted
                             WHERE _airbyte_loaded_at IS NULL
                             """.trimIndent()
+                        )
                     )
-                )
-                .iterateAll()
-                .iterator()
-                .next()
-                .first()
+                    .iterateAll()
+                    .iterator()
+                    .next()
+                    .first()
+            }
         // If this value is null, then there are no records with null loaded_at.
         // If it's not null, then we can return immediately - we've found some unprocessed records
         // and their timestamp.
@@ -52,18 +57,20 @@ class BigqueryTypingDedupingDatabaseInitialStatusGatherer(private val bq: BigQue
         }
 
         val loadedRecordTimestamp =
-            bq.query(
-                    QueryJobConfiguration.of(
-                        """
+            BigQueryUtils.executeBigQueryOperation {
+                bq.query(
+                        QueryJobConfiguration.of(
+                            """
                     SELECT MAX(_airbyte_extracted_at)
                     FROM $rawTableIdQuoted
                     """.trimIndent()
+                        )
                     )
-                )
-                .iterateAll()
-                .iterator()
-                .next()
-                .first()
+                    .iterateAll()
+                    .iterator()
+                    .next()
+                    .first()
+            }
         // We know (from the previous query) that all records have been processed by T+D already.
         // So we just need to get the timestamp of the most recent record.
         return if (loadedRecordTimestamp.isNull) {

airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/BigQueryUtilsExecuteBigQueryOperationTest.kt

@@ -0,0 +1,86 @@
+/*
+ * Copyright (c) 2026 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.bigquery
+
+import com.google.cloud.bigquery.BigQueryException
+import io.airbyte.cdk.TransientErrorException
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertFalse
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.assertThrows
+
+class BigQueryUtilsExecuteBigQueryOperationTest {
+
+    @Test
+    fun `successful operation returns result`() {
+        val result = BigQueryUtils.executeBigQueryOperation { "success" }
+        assertEquals("success", result)
+    }
+
+    @Test
+    fun `BigQueryException wrapping InterruptedException throws TransientErrorException`() {
+        val interruptedException = InterruptedException("thread interrupted")
+        val bigQueryException = BigQueryException(0, "interrupted", interruptedException)
+
+        val thrown =
+            assertThrows<TransientErrorException> {
+                BigQueryUtils.executeBigQueryOperation { throw bigQueryException }
+            }
+
+        assertEquals("BigQuery API call interrupted.", thrown.message)
+        assertEquals(bigQueryException, thrown.cause)
+        assertTrue(Thread.currentThread().isInterrupted)
+        // Clear the interrupted status for other tests
+        Thread.interrupted()
+    }
+
+    @Test
+    fun `BigQueryException wrapping InterruptedException restores interrupted status`() {
+        assertFalse(Thread.currentThread().isInterrupted)
+
+        val interruptedException = InterruptedException("thread interrupted")
+        val bigQueryException = BigQueryException(0, "interrupted", interruptedException)
+
+        assertThrows<TransientErrorException> {
+            BigQueryUtils.executeBigQueryOperation { throw bigQueryException }
+        }
+
+        assertTrue(Thread.currentThread().isInterrupted)
+        // Clear the interrupted status for other tests
+        Thread.interrupted()
+    }
+
+    @Test
+    fun `BigQueryException without InterruptedException cause is rethrown as-is`() {
+        val bigQueryException = BigQueryException(404, "not found")
+
+        val thrown =
+            assertThrows<BigQueryException> {
+                BigQueryUtils.executeBigQueryOperation { throw bigQueryException }
+            }
+
+        assertEquals(bigQueryException, thrown)
+    }
+
+    @Test
+    fun `BigQueryException with non-InterruptedException cause is rethrown as-is`() {
+        val ioException = java.io.IOException("network error")
+        val bigQueryException = BigQueryException(500, "server error", ioException)
+
+        val thrown =
+            assertThrows<BigQueryException> {
+                BigQueryUtils.executeBigQueryOperation { throw bigQueryException }
+            }
+
+        assertEquals(bigQueryException, thrown)
+    }
+
+    @Test
+    fun `null result from operation is returned`() {
+        val result: String? = BigQueryUtils.executeBigQueryOperation { null }
+        assertEquals(null, result)
+    }
+}

airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/typing_deduping/direct_load_tables/BigqueryDirectLoadDatabaseInitialStatusGathererTest.kt

@@ -0,0 +1,104 @@
+/*
+ * Copyright (c) 2026 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.bigquery.typing_deduping.direct_load_tables
+
+import com.google.cloud.bigquery.BigQuery
+import com.google.cloud.bigquery.BigQueryException
+import com.google.cloud.bigquery.Table
+import com.google.cloud.bigquery.TableId
+import io.airbyte.cdk.TransientErrorException
+import io.airbyte.cdk.load.command.Append
+import io.airbyte.cdk.load.command.DestinationStream
+import io.airbyte.cdk.load.command.NamespaceMapper
+import io.airbyte.cdk.load.data.ObjectType
+import io.airbyte.cdk.load.orchestration.db.ColumnNameMapping
+import io.airbyte.cdk.load.orchestration.db.TableName
+import io.airbyte.cdk.load.orchestration.db.TableNames
+import io.airbyte.cdk.load.orchestration.db.TempTableNameGenerator
+import io.airbyte.cdk.load.orchestration.db.legacy_typing_deduping.TableCatalog
+import io.airbyte.cdk.load.orchestration.db.legacy_typing_deduping.TableNameInfo
+import io.airbyte.integrations.destination.bigquery.write.typing_deduping.direct_load_tables.BigqueryDirectLoadDatabaseInitialStatusGatherer
+import io.mockk.every
+import io.mockk.mockk
+import java.math.BigInteger
+import kotlinx.coroutines.runBlocking
+import org.junit.jupiter.api.Assertions.assertNotNull
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.assertThrows
+
+class BigqueryDirectLoadDatabaseInitialStatusGathererTest {
+
+    private val bigquery: BigQuery = mockk()
+    private val tempTableNameGenerator: TempTableNameGenerator = mockk()
+
+    private val gatherer =
+        BigqueryDirectLoadDatabaseInitialStatusGatherer(bigquery, tempTableNameGenerator)
+
+    private val stream =
+        DestinationStream(
+            "test_namespace",
+            "test_stream",
+            Append,
+            ObjectType(linkedMapOf()),
+            generationId = 0,
+            minimumGenerationId = 0,
+            syncId = 0,
+            namespaceMapper = NamespaceMapper(),
+        )
+
+    private val tableName = TableName("test_namespace", "test_stream")
+    private val tempTableName = TableName("test_namespace", "test_stream_tmp")
+
+    @Test
+    fun `InterruptedException during getTable is wrapped as TransientErrorException`() =
+        runBlocking {
+            val interruptedException = InterruptedException("thread interrupted")
+            val bigQueryException = BigQueryException(0, "interrupted", interruptedException)
+
+            every { bigquery.getTable(any<TableId>()) } throws bigQueryException
+            every { tempTableNameGenerator.generate(any()) } returns tempTableName
+
+            val tableNames = TableNames(finalTableName = tableName, rawTableName = null)
+            val catalog =
+                TableCatalog(
+                    mapOf(stream to TableNameInfo(tableNames, ColumnNameMapping(emptyMap())))
+                )
+
+            assertThrows<TransientErrorException> { gatherer.gatherInitialStatus(catalog) }
+
+            // Clear interrupted status
+            Thread.interrupted()
+        }
+
+    @Test
+    fun `successful getTable returns status normally`() = runBlocking {
+        val table: Table = mockk()
+        every { table.numRows } returns BigInteger.ZERO
+
+        every { bigquery.getTable(any<TableId>()) } returns table
+        every { tempTableNameGenerator.generate(any()) } returns tempTableName
+
+        val tableNames = TableNames(finalTableName = tableName, rawTableName = null)
+        val catalog =
+            TableCatalog(mapOf(stream to TableNameInfo(tableNames, ColumnNameMapping(emptyMap()))))
+
+        val result = gatherer.gatherInitialStatus(catalog)
+        assertNotNull(result[stream])
+    }
+
+    @Test
+    fun `non-InterruptedException BigQueryException is rethrown as-is`() = runBlocking {
+        val bigQueryException = BigQueryException(403, "permission denied")
+
+        every { bigquery.getTable(any<TableId>()) } throws bigQueryException
+        every { tempTableNameGenerator.generate(any()) } returns tempTableName
+
+        val tableNames = TableNames(finalTableName = tableName, rawTableName = null)
+        val catalog =
+            TableCatalog(mapOf(stream to TableNameInfo(tableNames, ColumnNameMapping(emptyMap()))))
+
+        assertThrows<BigQueryException> { gatherer.gatherInitialStatus(catalog) }
+    }
+}