fix(destination-bigquery): surface interrupted standard-inserts operations as transient errors

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.23
+  dockerImageTag: 3.0.24-rc.1
   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,26 @@ object BigQueryUtils {
             Optional.ofNullable(System.getenv("WORKER_CONNECTOR_IMAGE"))
                 .map { name: String -> name.replace("airbyte/", "").replace(":", "/") }
                 .orElse("destination-bigquery")
+
+    @JvmStatic
+    inline fun <T> executeBigQueryOperation(operation: () -> T): T {
+        try {
+            return operation()
+        } catch (e: BigQueryException) {
+            var cause: Throwable? = e
+            while (cause != null) {
+                if (cause is InterruptedException) {
+                    Thread.currentThread().interrupt()
+                    throw TransientErrorException(
+                        "The BigQuery operation was interrupted, likely because the sync was cancelled. This is transient and the next sync attempt should succeed.",
+                        e,
+                    )
+                }
+                cause = cause.cause
+            }
+            throw e
+        }
+    }
 }
 
 fun TableId.toPrettyString() = "${this.dataset}.${this.table}"

airbyte-integrations/connectors/destination-bigquery/src/main/kotlin/io/airbyte/integrations/destination/bigquery/write/standard_insert/BigqueryBatchStandardInsertLoader.kt

@@ -76,7 +76,7 @@ class BigqueryBatchStandardInsertsLoader(
             "$formattedRecord${System.lineSeparator()}".toByteArray(StandardCharsets.UTF_8)
 
         if (this::writer.isInitialized) {
-            writer.write(ByteBuffer.wrap(byteArray))
+            BigQueryUtils.executeBigQueryOperation { writer.write(ByteBuffer.wrap(byteArray)) }
         } else {
             buffer!!.write(byteArray)
             // the default chunk size on the TableDataWriteChannel is 15MB,
@@ -94,9 +94,12 @@ class BigqueryBatchStandardInsertsLoader(
         if (!this::writer.isInitialized) {
             switchToWriteChannel()
         }
-        writer.close()
-        BigQueryUtils.waitForJobFinish(writer.job)
-        val stats = writer.job.reload().getStatistics<JobStatistics.LoadStatistics>()
+        BigQueryUtils.executeBigQueryOperation { writer.close() }
+        BigQueryUtils.executeBigQueryOperation { BigQueryUtils.waitForJobFinish(writer.job) }
+        val stats =
+            BigQueryUtils.executeBigQueryOperation {
+                writer.job.reload().getStatistics<JobStatistics.LoadStatistics>()
+            }
         logger.info {
             "Finished loading data into table ${writeChannelConfiguration.destinationTable.toPrettyString()}. ${stats.outputRows} rows loaded; ${stats.badRecords} bad records."
         }
@@ -117,18 +120,20 @@ class BigqueryBatchStandardInsertsLoader(
     private fun switchToWriteChannel() {
         writer =
             try {
-                bigquery.writer(job, writeChannelConfiguration)
+                BigQueryUtils.executeBigQueryOperation {
+                    bigquery.writer(job, writeChannelConfiguration)
+                }
             } catch (e: BigQueryException) {
                 if (e.code == HTTP_STATUS_CODE_FORBIDDEN || e.code == HTTP_STATUS_CODE_NOT_FOUND) {
                     throw ConfigErrorException(CONFIG_ERROR_MSG + e)
                 } else {
-                    throw BigQueryException(e.code, e.message)
+                    throw BigQueryException(e.code, e.message, e)
                 }
             }
         val byteArray = buffer!!.toByteArray()
         // please GC this object :)
         buffer = null
-        writer.write(ByteBuffer.wrap(byteArray))
+        BigQueryUtils.executeBigQueryOperation { writer.write(ByteBuffer.wrap(byteArray)) }
     }
 }
 

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

@@ -0,0 +1,48 @@
+/*
+ * 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.assertTrue
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.assertThrows
+
+class BigQueryUtilsExecuteBigQueryOperationTest {
+
+    @Test
+    fun `interrupted exception in cause chain becomes transient error`() {
+        val interrupted = InterruptedException("thread interrupted")
+        val nestedCause = IllegalStateException("nested", interrupted)
+        val bigQueryException = BigQueryException(500, "operation failed", nestedCause)
+
+        val thrown =
+            assertThrows<TransientErrorException> {
+                BigQueryUtils.executeBigQueryOperation<Nothing> { throw bigQueryException }
+            }
+
+        assertEquals(
+            "The BigQuery operation was interrupted, likely because the sync was cancelled. This is transient and the next sync attempt should succeed.",
+            thrown.message,
+        )
+        assertEquals(bigQueryException, thrown.cause)
+        assertTrue(Thread.currentThread().isInterrupted)
+        Thread.interrupted()
+    }
+
+    @Test
+    fun `non-interrupted BigQuery exception is rethrown`() {
+        val bigQueryException =
+            BigQueryException(500, "operation failed", RuntimeException("cause"))
+
+        val thrown =
+            assertThrows<BigQueryException> {
+                BigQueryUtils.executeBigQueryOperation<Nothing> { throw bigQueryException }
+            }
+
+        assertEquals(bigQueryException, thrown)
+    }
+}

airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/write/standard_insert/BigqueryBatchStandardInsertsLoaderTest.kt

@@ -0,0 +1,97 @@
+/*
+ * Copyright (c) 2026 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.bigquery.write.standard_insert
+
+import com.google.cloud.bigquery.BigQuery
+import com.google.cloud.bigquery.BigQueryException
+import com.google.cloud.bigquery.JobId
+import com.google.cloud.bigquery.JobInfo
+import com.google.cloud.bigquery.TableId
+import com.google.cloud.bigquery.WriteChannelConfiguration
+import io.airbyte.cdk.ConfigErrorException
+import io.airbyte.cdk.TransientErrorException
+import io.airbyte.cdk.load.message.DestinationRecordRaw
+import io.mockk.every
+import io.mockk.mockk
+import kotlinx.coroutines.runBlocking
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.Assertions.assertTrue
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.assertThrows
+import org.junit.jupiter.params.ParameterizedTest
+import org.junit.jupiter.params.provider.ValueSource
+
+class BigqueryBatchStandardInsertsLoaderTest {
+
+    private val oversizedRecord = "record".repeat(3 * 1024 * 1024)
+    private val bigquery: BigQuery = mockk()
+    private val formatter: RecordFormatter = mockk()
+    private val jobId = JobId.newBuilder().setRandomJob().build()
+    private val configuration =
+        WriteChannelConfiguration.newBuilder(TableId.of("dataset", "table"))
+            .setCreateDisposition(JobInfo.CreateDisposition.CREATE_IF_NEEDED)
+            .build()
+
+    @ParameterizedTest
+    @ValueSource(ints = [403, 404])
+    fun `forbidden and not found writer errors remain config errors`(code: Int) {
+        val exception = BigQueryException(code, "permission denied")
+        every { bigquery.writer(any<JobId>(), any()) } throws exception
+        every { formatter.formatRecord(any<DestinationRecordRaw>()) } returns oversizedRecord
+
+        val loader = loader()
+        val thrown = assertThrows<ConfigErrorException> { runBlocking { loader.accept(record()) } }
+
+        assertEquals(
+            BigqueryBatchStandardInsertsLoaderFactory.CONFIG_ERROR_MSG + exception,
+            thrown.message,
+        )
+    }
+
+    @Test
+    fun `writer error preserves its cause`() {
+        val cause = IllegalStateException("writer failed")
+        val exception = BigQueryException(500, "operation failed", cause)
+        every { bigquery.writer(any<JobId>(), any()) } throws exception
+        every { formatter.formatRecord(any<DestinationRecordRaw>()) } returns oversizedRecord
+
+        val loader = loader()
+        val thrown = assertThrows<BigQueryException> { runBlocking { loader.accept(record()) } }
+
+        assertEquals(exception.code, thrown.code)
+        assertEquals(exception.message, thrown.message)
+        assertEquals(exception, thrown.cause)
+    }
+
+    @Test
+    fun `interrupted writer error becomes transient error and restores interrupt flag`() {
+        Thread.interrupted()
+        val interrupted = InterruptedException("thread interrupted")
+        val exception =
+            BigQueryException(
+                500,
+                "operation interrupted",
+                IllegalStateException("nested cause", interrupted),
+            )
+        every { bigquery.writer(any<JobId>(), any()) } throws exception
+        every { formatter.formatRecord(any<DestinationRecordRaw>()) } returns oversizedRecord
+
+        try {
+            val loader = loader()
+            val thrown =
+                assertThrows<TransientErrorException> { runBlocking { loader.accept(record()) } }
+
+            assertEquals(exception, thrown.cause)
+            assertTrue(Thread.currentThread().isInterrupted)
+        } finally {
+            Thread.interrupted()
+        }
+    }
+
+    private fun loader() =
+        BigqueryBatchStandardInsertsLoader(bigquery, configuration, jobId, formatter)
+
+    private fun record(): DestinationRecordRaw = mockk()
+}

docs/integrations/destinations/bigquery.md

@@ -252,6 +252,7 @@ This destination supports [namespaces](https://docs.airbyte.com/platform/using-a
 
 | Version     | Date       | Pull Request                                               | Subject                                                                                                                                                                           |
 |:------------|:-----------|:-----------------------------------------------------------|:----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| 3.0.24-rc.1 | 2026-07-31 | [83276](https://github.com/airbytehq/airbyte/pull/83276) | Surface interrupted BigQuery operations as transient errors instead of system errors during sync teardown |
 | 3.0.23 | 2026-07-14 | [81550](https://github.com/airbytehq/airbyte/pull/81550) | Use CREATE TABLE IF NOT EXISTS for non-replace table creation to prevent accidental data loss |
 | 3.0.22 | 2026-07-10 | [81635](https://github.com/airbytehq/airbyte/pull/81635) | Restore PK NULL equality checks |
 | 3.0.21 | 2026-06-30 | [81346](https://github.com/airbytehq/airbyte/pull/81346) | Remove unnecessary NULL PK equality checks from merge SQL |