From 9247abe3e028a43467cd0ecb9590ebef7d4fa743 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 31 Jul 2026 03:32:27 +0000 Subject: [PATCH 1/4] fix(destination-bigquery): handle interrupted standard inserts Co-Authored-By: bot_apk --- .../destination-bigquery/metadata.yaml | 2 +- .../destination/bigquery/BigQueryUtils.kt | 21 ++++++ .../BigqueryBatchStandardInsertLoader.kt | 23 ++++-- ...gQueryUtilsExecuteBigQueryOperationTest.kt | 47 ++++++++++++ .../BigqueryBatchStandardInsertsLoaderTest.kt | 75 +++++++++++++++++++ docs/integrations/destinations/bigquery.md | 1 + 6 files changed, 161 insertions(+), 8 deletions(-) create mode 100644 airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/BigQueryUtilsExecuteBigQueryOperationTest.kt create mode 100644 airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/write/standard_insert/BigqueryBatchStandardInsertsLoaderTest.kt diff --git a/airbyte-integrations/connectors/destination-bigquery/metadata.yaml b/airbyte-integrations/connectors/destination-bigquery/metadata.yaml index dfa22a1ad6b5..03fec47554b6 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.23 + dockerImageTag: 3.0.24-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/BigQueryUtils.kt b/airbyte-integrations/connectors/destination-bigquery/src/main/kotlin/io/airbyte/integrations/destination/bigquery/BigQueryUtils.kt index 7a473c55d48e..35a5adcc8c58 100644 --- a/airbyte-integrations/connectors/destination-bigquery/src/main/kotlin/io/airbyte/integrations/destination/bigquery/BigQueryUtils.kt +++ b/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 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}" 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 54f34c4cfb67..cc9eebc52257 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 @@ -76,7 +76,9 @@ 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 +96,12 @@ class BigqueryBatchStandardInsertsLoader( if (!this::writer.isInitialized) { switchToWriteChannel() } - writer.close() - BigQueryUtils.waitForJobFinish(writer.job) - val stats = writer.job.reload().getStatistics() + BigQueryUtils.executeBigQueryOperation { writer.close() } + BigQueryUtils.executeBigQueryOperation { BigQueryUtils.waitForJobFinish(writer.job) } + val stats = + BigQueryUtils.executeBigQueryOperation { + writer.job.reload().getStatistics() + } logger.info { "Finished loading data into table ${writeChannelConfiguration.destinationTable.toPrettyString()}. ${stats.outputRows} rows loaded; ${stats.badRecords} bad records." } @@ -117,18 +122,22 @@ 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)) + } } } diff --git a/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/BigQueryUtilsExecuteBigQueryOperationTest.kt b/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/BigQueryUtilsExecuteBigQueryOperationTest.kt new file mode 100644 index 000000000000..0a566285c0d3 --- /dev/null +++ b/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/BigQueryUtilsExecuteBigQueryOperationTest.kt @@ -0,0 +1,47 @@ +/* + * 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 { + BigQueryUtils.executeBigQueryOperation { 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 { + BigQueryUtils.executeBigQueryOperation { throw bigQueryException } + } + + assertEquals(bigQueryException, thrown) + } +} 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 new file mode 100644 index 000000000000..c06bc057c966 --- /dev/null +++ b/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/write/standard_insert/BigqueryBatchStandardInsertsLoaderTest.kt @@ -0,0 +1,75 @@ +/* + * 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.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.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(), any()) } throws exception + every { formatter.formatRecord(any()) } returns oversizedRecord + + val loader = loader() + val thrown = + assertThrows { + 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(), any()) } throws exception + every { formatter.formatRecord(any()) } returns oversizedRecord + + val loader = loader() + val thrown = + assertThrows { + runBlocking { loader.accept(record()) } + } + + assertEquals(exception.code, thrown.code) + assertEquals(exception.message, thrown.message) + assertEquals(exception, thrown.cause) + } + + private fun loader() = + BigqueryBatchStandardInsertsLoader(bigquery, configuration, jobId, formatter) + + private fun record(): DestinationRecordRaw = mockk() +} diff --git a/docs/integrations/destinations/bigquery.md b/docs/integrations/destinations/bigquery.md index 68125e3e9dbd..5ee65b55b8b6 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.24-rc.1 | 2026-07-31 | [00000](https://github.com/airbytehq/airbyte/pull/00000) | 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 | From 039a2b1038b7cd4353e1f4d4210e8c6cf6cbb0af Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 31 Jul 2026 03:38:00 +0000 Subject: [PATCH 2/4] test(destination-bigquery): cover interrupted standard inserts Co-Authored-By: bot_apk --- .../BigqueryBatchStandardInsertsLoaderTest.kt | 30 +++++++++++++++++++ 1 file changed, 30 insertions(+) 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 c06bc057c966..592ccf3e5bc0 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 @@ -11,11 +11,14 @@ 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 @@ -68,6 +71,33 @@ class BigqueryBatchStandardInsertsLoaderTest { 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(), any()) } throws exception + every { formatter.formatRecord(any()) } returns oversizedRecord + + try { + val loader = loader() + val thrown = + assertThrows { + runBlocking { loader.accept(record()) } + } + + assertEquals(exception, thrown.cause) + assertTrue(Thread.currentThread().isInterrupted) + } finally { + Thread.interrupted() + } + } + private fun loader() = BigqueryBatchStandardInsertsLoader(bigquery, configuration, jobId, formatter) From 1fea16899aae9a88fb1c22486c80ae8672e23e1a Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 31 Jul 2026 03:39:39 +0000 Subject: [PATCH 3/4] docs(destination-bigquery): set changelog PR number Co-Authored-By: bot_apk --- docs/integrations/destinations/bigquery.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/integrations/destinations/bigquery.md b/docs/integrations/destinations/bigquery.md index 5ee65b55b8b6..8037143738bd 100644 --- a/docs/integrations/destinations/bigquery.md +++ b/docs/integrations/destinations/bigquery.md @@ -252,7 +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 | [00000](https://github.com/airbytehq/airbyte/pull/00000) | Surface interrupted BigQuery operations as transient errors instead of system errors during sync teardown | +| 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 | From 60ac12c70471c3cb108a109e4b548a5d11622ca8 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 31 Jul 2026 03:47:30 +0000 Subject: [PATCH 4/4] style(destination-bigquery): apply spotless formatting Co-Authored-By: bot_apk --- .../BigqueryBatchStandardInsertLoader.kt | 8 ++------ .../BigQueryUtilsExecuteBigQueryOperationTest.kt | 3 ++- .../BigqueryBatchStandardInsertsLoaderTest.kt | 14 +++----------- 3 files changed, 7 insertions(+), 18 deletions(-) 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 cc9eebc52257..6d448e70fedd 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 @@ -76,9 +76,7 @@ class BigqueryBatchStandardInsertsLoader( "$formattedRecord${System.lineSeparator()}".toByteArray(StandardCharsets.UTF_8) if (this::writer.isInitialized) { - BigQueryUtils.executeBigQueryOperation { - 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, @@ -135,9 +133,7 @@ class BigqueryBatchStandardInsertsLoader( val byteArray = buffer!!.toByteArray() // please GC this object :) buffer = null - BigQueryUtils.executeBigQueryOperation { - writer.write(ByteBuffer.wrap(byteArray)) - } + BigQueryUtils.executeBigQueryOperation { writer.write(ByteBuffer.wrap(byteArray)) } } } diff --git a/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/BigQueryUtilsExecuteBigQueryOperationTest.kt b/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/BigQueryUtilsExecuteBigQueryOperationTest.kt index 0a566285c0d3..9a5b61b4afe7 100644 --- a/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/BigQueryUtilsExecuteBigQueryOperationTest.kt +++ b/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/BigQueryUtilsExecuteBigQueryOperationTest.kt @@ -35,7 +35,8 @@ class BigQueryUtilsExecuteBigQueryOperationTest { @Test fun `non-interrupted BigQuery exception is rethrown`() { - val bigQueryException = BigQueryException(500, "operation failed", RuntimeException("cause")) + val bigQueryException = + BigQueryException(500, "operation failed", RuntimeException("cause")) val thrown = assertThrows { 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 592ccf3e5bc0..54ed8f8a3af3 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 @@ -42,10 +42,7 @@ class BigqueryBatchStandardInsertsLoaderTest { every { formatter.formatRecord(any()) } returns oversizedRecord val loader = loader() - val thrown = - assertThrows { - runBlocking { loader.accept(record()) } - } + val thrown = assertThrows { runBlocking { loader.accept(record()) } } assertEquals( BigqueryBatchStandardInsertsLoaderFactory.CONFIG_ERROR_MSG + exception, @@ -61,10 +58,7 @@ class BigqueryBatchStandardInsertsLoaderTest { every { formatter.formatRecord(any()) } returns oversizedRecord val loader = loader() - val thrown = - assertThrows { - runBlocking { loader.accept(record()) } - } + val thrown = assertThrows { runBlocking { loader.accept(record()) } } assertEquals(exception.code, thrown.code) assertEquals(exception.message, thrown.message) @@ -87,9 +81,7 @@ class BigqueryBatchStandardInsertsLoaderTest { try { val loader = loader() val thrown = - assertThrows { - runBlocking { loader.accept(record()) } - } + assertThrows { runBlocking { loader.accept(record()) } } assertEquals(exception, thrown.cause) assertTrue(Thread.currentThread().isInterrupted)