diff --git a/airbyte-integrations/connectors/destination-bigquery/metadata.yaml b/airbyte-integrations/connectors/destination-bigquery/metadata.yaml index 03fec47554b6..aad7ee36dcd9 100644 --- a/airbyte-integrations/connectors/destination-bigquery/metadata.yaml +++ b/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 diff --git a/airbyte-integrations/connectors/destination-bigquery/src/main/kotlin/io/airbyte/integrations/destination/bigquery/write/standard_insert/BigqueryBatchStandardInsertLoader.kt b/airbyte-integrations/connectors/destination-bigquery/src/main/kotlin/io/airbyte/integrations/destination/bigquery/write/standard_insert/BigqueryBatchStandardInsertLoader.kt index 6d448e70fedd..3b1ca012f9ad 100644 --- a/airbyte-integrations/connectors/destination-bigquery/src/main/kotlin/io/airbyte/integrations/destination/bigquery/write/standard_insert/BigqueryBatchStandardInsertLoader.kt +++ b/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 { diff --git a/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/write/standard_insert/BigqueryBatchStandardInsertsLoaderTest.kt b/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/write/standard_insert/BigqueryBatchStandardInsertsLoaderTest.kt index 54ed8f8a3af3..03fd748eb35d 100644 --- a/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/write/standard_insert/BigqueryBatchStandardInsertsLoaderTest.kt +++ b/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/write/standard_insert/BigqueryBatchStandardInsertsLoaderTest.kt @@ -8,6 +8,7 @@ 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 @@ -15,6 +16,7 @@ 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(), any()) } } @Test - fun `writer error preserves its cause`() { + fun `retryable writer errors retry and then succeed`() { + val writer = mockk(relaxed = true) + every { bigquery.writer(any(), any()) } throws + BigQueryException(503, "backend unavailable") andThenThrows + BigQueryException(503, "backend unavailable") andThen + writer + every { formatter.formatRecord(any()) } returns oversizedRecord + + val loader = loader(maxOpenAttempts = 5) + runBlocking { loader.accept(record()) } + + verify(exactly = 3) { bigquery.writer(any(), any()) } + } + + @Test + fun `writer retries are exhausted as transient error`() { + val exception = BigQueryException(503, "backend unavailable") + every { bigquery.writer(any(), any()) } throws exception + every { formatter.formatRecord(any()) } returns oversizedRecord + + val loader = loader(maxOpenAttempts = 4) + val thrown = + assertThrows { runBlocking { loader.accept(record()) } } + + assertEquals(exception, thrown.cause) + verify(exactly = 4) { bigquery.writer(any(), 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(), any()) } throws exception every { formatter.formatRecord(any()) } 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(), 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() } diff --git a/docs/integrations/destinations/bigquery.md b/docs/integrations/destinations/bigquery.md index 8037143738bd..9faeb17081f1 100644 --- a/docs/integrations/destinations/bigquery.md +++ b/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 |