fix(destination-bigquery): retry transient backend errors when opening write channel

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.24-rc.1
+  dockerImageTag: 3.0.25-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/write/standard_insert/BigqueryBatchStandardInsertLoader.kt

@@ -16,6 +16,7 @@ import com.google.cloud.bigquery.TableId
 import com.google.cloud.bigquery.WriteChannelConfiguration
 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings
 import io.airbyte.cdk.ConfigErrorException
+import io.airbyte.cdk.TransientErrorException
 import io.airbyte.cdk.load.command.DestinationCatalog
 import io.airbyte.cdk.load.command.DestinationStream
 import io.airbyte.cdk.load.config.DataChannelFormat
@@ -45,6 +46,8 @@ import jakarta.inject.Singleton
 import java.io.ByteArrayOutputStream
 import java.nio.ByteBuffer
 import java.nio.charset.StandardCharsets
+import kotlin.math.min
+import kotlin.random.Random
 
 private val logger = KotlinLogging.logger {}
 
@@ -57,6 +60,11 @@ class BigqueryBatchStandardInsertsLoader(
     private val writeChannelConfiguration: WriteChannelConfiguration,
     private val job: JobId,
     private val recordFormatter: RecordFormatter,
+    private val sleep: (Long) -> Unit = { Thread.sleep(it) },
+    private val maxOpenAttempts: Int = MAX_OPEN_ATTEMPTS,
+    private val initialOpenDelayMs: Long = INITIAL_OPEN_DELAY_MS,
+    private val maxOpenDelayMs: Long = MAX_OPEN_DELAY_MS,
+    private val jitterMs: () -> Long = { Random.nextLong(0, MAX_JITTER_MS + 1) },
 ) : DirectLoader {
     // a TableDataWriteChannel holds (by default) a 15MB buffer in memory.
     // so we start out by writing to a BAOS, which grows dynamically.
@@ -118,23 +126,47 @@ class BigqueryBatchStandardInsertsLoader(
     // check...
     @SuppressFBWarnings(value = ["RCN_REDUNDANT_NULLCHECK_WOULD_HAVE_BEEN_A_NPE"])
     private fun switchToWriteChannel() {
-        writer =
+        var delayMs = initialOpenDelayMs
+        for (attempt in 1..maxOpenAttempts) {
             try {
-                BigQueryUtils.executeBigQueryOperation {
-                    bigquery.writer(job, writeChannelConfiguration)
-                }
+                writer =
+                    BigQueryUtils.executeBigQueryOperation {
+                        bigquery.writer(job, writeChannelConfiguration)
+                    }
+                break
             } catch (e: BigQueryException) {
                 if (e.code == HTTP_STATUS_CODE_FORBIDDEN || e.code == HTTP_STATUS_CODE_NOT_FOUND) {
                     throw ConfigErrorException(CONFIG_ERROR_MSG + e)
-                } else {
+                }
+                if (e.code !in RETRYABLE_BACKEND_STATUS_CODES) {
                     throw BigQueryException(e.code, e.message, e)
                 }
+                if (attempt == maxOpenAttempts) {
+                    throw TransientErrorException(
+                        "The BigQuery backend was unavailable while opening the standard-inserts write channel. The next sync attempt should succeed.",
+                        e,
+                    )
+                }
+                logger.warn(e) {
+                    "Retrying BigQuery write-channel open (attempt ${attempt + 1}/$maxOpenAttempts, HTTP ${e.code}): ${e.message}"
+                }
+                sleep(delayMs + jitterMs())
+                delayMs = min(delayMs * 2, maxOpenDelayMs)
             }
+        }
         val byteArray = buffer!!.toByteArray()
         // please GC this object :)
         buffer = null
         BigQueryUtils.executeBigQueryOperation { writer.write(ByteBuffer.wrap(byteArray)) }
     }
+
+    companion object {
+        internal const val MAX_OPEN_ATTEMPTS = 5
+        internal const val INITIAL_OPEN_DELAY_MS = 1000L
+        internal const val MAX_OPEN_DELAY_MS = 30_000L
+        internal const val MAX_JITTER_MS = 1000L
+        private val RETRYABLE_BACKEND_STATUS_CODES = setOf(500, 502, 503, 504)
+    }
 }
 
 class BigqueryConfiguredForBatchStandardInserts : Condition {

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

@@ -8,13 +8,15 @@ 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.TableDataWriteChannel
 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 io.mockk.verify
 import kotlinx.coroutines.runBlocking
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Assertions.assertTrue
@@ -48,12 +50,42 @@ class BigqueryBatchStandardInsertsLoaderTest {
             BigqueryBatchStandardInsertsLoaderFactory.CONFIG_ERROR_MSG + exception,
             thrown.message,
         )
+        verify(exactly = 1) { bigquery.writer(any<JobId>(), any()) }
     }
 
     @Test
-    fun `writer error preserves its cause`() {
+    fun `retryable writer errors retry and then succeed`() {
+        val writer = mockk<TableDataWriteChannel>(relaxed = true)
+        every { bigquery.writer(any<JobId>(), any()) } throws
+            BigQueryException(503, "backend unavailable") andThenThrows
+            BigQueryException(503, "backend unavailable") andThen
+            writer
+        every { formatter.formatRecord(any<DestinationRecordRaw>()) } returns oversizedRecord
+
+        val loader = loader(maxOpenAttempts = 5)
+        runBlocking { loader.accept(record()) }
+
+        verify(exactly = 3) { bigquery.writer(any<JobId>(), any()) }
+    }
+
+    @Test
+    fun `writer retries are exhausted as transient error`() {
+        val exception = BigQueryException(503, "backend unavailable")
+        every { bigquery.writer(any<JobId>(), any()) } throws exception
+        every { formatter.formatRecord(any<DestinationRecordRaw>()) } returns oversizedRecord
+
+        val loader = loader(maxOpenAttempts = 4)
+        val thrown =
+            assertThrows<TransientErrorException> { runBlocking { loader.accept(record()) } }
+
+        assertEquals(exception, thrown.cause)
+        verify(exactly = 4) { bigquery.writer(any<JobId>(), any()) }
+    }
+
+    @Test
+    fun `non-retryable writer error preserves its cause`() {
         val cause = IllegalStateException("writer failed")
-        val exception = BigQueryException(500, "operation failed", cause)
+        val exception = BigQueryException(400, "operation failed", cause)
         every { bigquery.writer(any<JobId>(), any()) } throws exception
         every { formatter.formatRecord(any<DestinationRecordRaw>()) } returns oversizedRecord
 
@@ -63,6 +95,7 @@ class BigqueryBatchStandardInsertsLoaderTest {
         assertEquals(exception.code, thrown.code)
         assertEquals(exception.message, thrown.message)
         assertEquals(exception, thrown.cause)
+        verify(exactly = 1) { bigquery.writer(any<JobId>(), any()) }
     }
 
     @Test
@@ -90,8 +123,18 @@ class BigqueryBatchStandardInsertsLoaderTest {
         }
     }
 
-    private fun loader() =
-        BigqueryBatchStandardInsertsLoader(bigquery, configuration, jobId, formatter)
+    private fun loader(
+        maxOpenAttempts: Int = BigqueryBatchStandardInsertsLoader.MAX_OPEN_ATTEMPTS
+    ) =
+        BigqueryBatchStandardInsertsLoader(
+            bigquery,
+            configuration,
+            jobId,
+            formatter,
+            sleep = {},
+            maxOpenAttempts = maxOpenAttempts,
+            jitterMs = { 0 },
+        )
 
     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.25-rc.1 | 2026-08-12 | [84296](https://github.com/airbytehq/airbyte/pull/84296) | Retry transient BigQuery backend errors when opening the standard-inserts write channel and surface exhausted retries as a transient error |
 | 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 |