From 20d255866a96e602e861daa83b2d64a8ea3f85a1 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 8 Jun 2026 11:46:25 +0000 Subject: [PATCH 1/4] fix(destination-bigquery): handle InterruptedException during sync cancellation Wrap BigQuery API calls in status gatherers with executeBigQueryOperation() to catch BigQueryException caused by InterruptedException. This converts the raw Java exception into a user-friendly TransientErrorException instead of surfacing 'java.lang.InterruptedException' as a system_error. Affected components: - BigqueryDirectLoadDatabaseInitialStatusGatherer - BigqueryTypingDedupingDatabaseInitialStatusGatherer - BigQueryUtils (new executeBigQueryOperation utility) Co-Authored-By: bot_apk --- .../destination-bigquery/metadata.yaml | 2 +- .../destination/bigquery/BigQueryUtils.kt | 23 ++++ ...DirectLoadDatabaseInitialStatusGatherer.kt | 4 +- ...ngDedupingDatabaseInitialStatusGatherer.kt | 41 ++++--- ...gQueryUtilsExecuteBigQueryOperationTest.kt | 86 ++++++++++++++ ...ctLoadDatabaseInitialStatusGathererTest.kt | 107 ++++++++++++++++++ ...dupingDatabaseInitialStatusGathererTest.kt | 98 ++++++++++++++++ docs/integrations/destinations/bigquery.md | 1 + 8 files changed, 343 insertions(+), 19 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/typing_deduping/direct_load_tables/BigqueryDirectLoadDatabaseInitialStatusGathererTest.kt create mode 100644 airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/typing_deduping/legacy_raw_tables/BigqueryTypingDedupingDatabaseInitialStatusGathererTest.kt diff --git a/airbyte-integrations/connectors/destination-bigquery/metadata.yaml b/airbyte-integrations/connectors/destination-bigquery/metadata.yaml index 1aae1ff26985..9ce3bbc570aa 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.19 + dockerImageTag: 3.0.20 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..3e6dbc664592 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 @@ -11,6 +11,7 @@ import com.google.common.collect.ImmutableList import com.google.common.collect.ImmutableMap import io.airbyte.cdk.ConfigErrorException import io.airbyte.cdk.load.message.Meta +import io.airbyte.commons.exceptions.TransientErrorException import java.util.* import java.util.stream.Collectors import org.apache.commons.lang3.StringUtils @@ -176,6 +177,28 @@ 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 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, + "BigQuery operation interrupted, likely due to sync cancellation: ${e.message}", + ) + } + 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/typing_deduping/direct_load_tables/BigqueryDirectLoadDatabaseInitialStatusGatherer.kt b/airbyte-integrations/connectors/destination-bigquery/src/main/kotlin/io/airbyte/integrations/destination/bigquery/write/typing_deduping/direct_load_tables/BigqueryDirectLoadDatabaseInitialStatusGatherer.kt index 0c896b997f80..1cf1f4414c15 100644 --- a/airbyte-integrations/connectors/destination-bigquery/src/main/kotlin/io/airbyte/integrations/destination/bigquery/write/typing_deduping/direct_load_tables/BigqueryDirectLoadDatabaseInitialStatusGatherer.kt +++ b/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) } } } diff --git a/airbyte-integrations/connectors/destination-bigquery/src/main/kotlin/io/airbyte/integrations/destination/bigquery/write/typing_deduping/legacy_raw_tables/BigqueryTypingDedupingDatabaseInitialStatusGatherer.kt b/airbyte-integrations/connectors/destination-bigquery/src/main/kotlin/io/airbyte/integrations/destination/bigquery/write/typing_deduping/legacy_raw_tables/BigqueryTypingDedupingDatabaseInitialStatusGatherer.kt index 4ae371809703..2b6756447f76 100644 --- a/airbyte-integrations/connectors/destination-bigquery/src/main/kotlin/io/airbyte/integrations/destination/bigquery/write/typing_deduping/legacy_raw_tables/BigqueryTypingDedupingDatabaseInitialStatusGatherer.kt +++ b/airbyte-integrations/connectors/destination-bigquery/src/main/kotlin/io/airbyte/integrations/destination/bigquery/write/typing_deduping/legacy_raw_tables/BigqueryTypingDedupingDatabaseInitialStatusGatherer.kt @@ -8,6 +8,7 @@ import com.google.cloud.bigquery.BigQuery import com.google.cloud.bigquery.QueryJobConfiguration import com.google.cloud.bigquery.TableId import io.airbyte.cdk.load.command.DestinationStream +import io.airbyte.integrations.destination.bigquery.BigQueryUtils import io.airbyte.cdk.load.orchestration.db.DatabaseInitialStatusGatherer import io.airbyte.cdk.load.orchestration.db.TableName import io.airbyte.cdk.load.orchestration.db.TableNames @@ -22,25 +23,29 @@ class BigqueryTypingDedupingDatabaseInitialStatusGatherer(private val bq: BigQue 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) { 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..9e2a6c75f6ac --- /dev/null +++ b/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.commons.exceptions.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 { + 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 { + 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 { + 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 { + BigQueryUtils.executeBigQueryOperation { throw bigQueryException } + } + + assertEquals(bigQueryException, thrown) + } + + @Test + fun `null result from operation is returned`() { + val result: String? = BigQueryUtils.executeBigQueryOperation { null } + assertEquals(null, result) + } +} diff --git a/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/typing_deduping/direct_load_tables/BigqueryDirectLoadDatabaseInitialStatusGathererTest.kt b/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/typing_deduping/direct_load_tables/BigqueryDirectLoadDatabaseInitialStatusGathererTest.kt new file mode 100644 index 000000000000..57957fd0a20f --- /dev/null +++ b/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/typing_deduping/direct_load_tables/BigqueryDirectLoadDatabaseInitialStatusGathererTest.kt @@ -0,0 +1,107 @@ +/* + * 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.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.commons.exceptions.TransientErrorException +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()) } 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 { 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()) } 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()) } 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 { gatherer.gatherInitialStatus(catalog) } + } +} diff --git a/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/typing_deduping/legacy_raw_tables/BigqueryTypingDedupingDatabaseInitialStatusGathererTest.kt b/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/typing_deduping/legacy_raw_tables/BigqueryTypingDedupingDatabaseInitialStatusGathererTest.kt new file mode 100644 index 000000000000..6a685c1e5ac6 --- /dev/null +++ b/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/typing_deduping/legacy_raw_tables/BigqueryTypingDedupingDatabaseInitialStatusGathererTest.kt @@ -0,0 +1,98 @@ +/* + * Copyright (c) 2026 Airbyte, Inc., all rights reserved. + */ + +package io.airbyte.integrations.destination.bigquery.typing_deduping.legacy_raw_tables + +import com.google.cloud.bigquery.BigQuery +import com.google.cloud.bigquery.BigQueryException +import com.google.cloud.bigquery.TableId +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.legacy_typing_deduping.TableCatalog +import io.airbyte.cdk.load.orchestration.db.legacy_typing_deduping.TableNameInfo +import io.airbyte.commons.exceptions.TransientErrorException +import io.airbyte.integrations.destination.bigquery.write.typing_deduping.legacy_raw_tables.BigqueryTypingDedupingDatabaseInitialStatusGatherer +import io.mockk.every +import io.mockk.mockk +import kotlinx.coroutines.runBlocking +import org.junit.jupiter.api.Assertions.assertNotNull +import org.junit.jupiter.api.Assertions.assertNull +import org.junit.jupiter.api.Test +import org.junit.jupiter.api.assertThrows + +class BigqueryTypingDedupingDatabaseInitialStatusGathererTest { + + private val bq: BigQuery = mockk() + + private val gatherer = BigqueryTypingDedupingDatabaseInitialStatusGatherer(bq) + + private val stream = + DestinationStream( + "test_namespace", + "test_stream", + Append, + ObjectType(linkedMapOf()), + generationId = 0, + minimumGenerationId = 0, + syncId = 0, + namespaceMapper = NamespaceMapper(), + ) + + private val rawTableName = TableName("test_namespace", "test_stream_raw") + private val finalTableName = TableName("test_namespace", "test_stream") + + @Test + fun `InterruptedException during getTable is wrapped as TransientErrorException`() = runBlocking { + val interruptedException = InterruptedException("thread interrupted") + val bigQueryException = BigQueryException(0, "interrupted", interruptedException) + + every { bq.getTable(any()) } throws bigQueryException + + val tableNames = TableNames(rawTableName = rawTableName, finalTableName = finalTableName) + val catalog = + TableCatalog( + mapOf(stream to TableNameInfo(tableNames, ColumnNameMapping(emptyMap()))) + ) + + assertThrows { gatherer.gatherInitialStatus(catalog) } + + // Clear interrupted status + Thread.interrupted() + } + + @Test + fun `table not found returns status with null raw table state`() = runBlocking { + every { bq.getTable(any()) } returns null + + val tableNames = TableNames(rawTableName = rawTableName, finalTableName = finalTableName) + val catalog = + TableCatalog( + mapOf(stream to TableNameInfo(tableNames, ColumnNameMapping(emptyMap()))) + ) + + val result = gatherer.gatherInitialStatus(catalog) + assertNotNull(result[stream]) + assertNull(result[stream]!!.rawTableStatus) + } + + @Test + fun `non-InterruptedException BigQueryException is rethrown as-is`() = runBlocking { + val bigQueryException = BigQueryException(403, "permission denied") + + every { bq.getTable(any()) } throws bigQueryException + + val tableNames = TableNames(rawTableName = rawTableName, finalTableName = finalTableName) + val catalog = + TableCatalog( + mapOf(stream to TableNameInfo(tableNames, ColumnNameMapping(emptyMap()))) + ) + + assertThrows { gatherer.gatherInitialStatus(catalog) } + } +} diff --git a/docs/integrations/destinations/bigquery.md b/docs/integrations/destinations/bigquery.md index 2306303044ff..d3f07603e6ca 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.20 | 2026-06-08 | [TBD](https://github.com/airbytehq/airbyte/pull/TBD) | Improved error handling for InterruptedException during sync cancellation. | | 3.0.19 | 2026-05-21 | [78239](https://github.com/airbytehq/airbyte/pull/78239) | Promoting release candidate 3.0.19-rc.1 to a main version. | | 3.0.19-rc.1 | 2026-05-19 | [78239](https://github.com/airbytehq/airbyte/pull/78239) | Upgrade CDK to 1.0.13. Progressive rollout. | | 3.0.18 | 2026-03-31 | [75913](https://github.com/airbytehq/airbyte/pull/75913) | Finalize upgrade BigQuery Cloud dependencies and CDK version | From 973f7fe23870c122df2089598993c85b1eacc311 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 8 Jun 2026 11:53:02 +0000 Subject: [PATCH 2/4] style: apply ktfmt formatting fixes Co-Authored-By: bot_apk --- ...ngDedupingDatabaseInitialStatusGatherer.kt | 2 +- ...ctLoadDatabaseInitialStatusGathererTest.kt | 37 +++++++++---------- ...dupingDatabaseInitialStatusGathererTest.kt | 36 +++++++++--------- 3 files changed, 35 insertions(+), 40 deletions(-) diff --git a/airbyte-integrations/connectors/destination-bigquery/src/main/kotlin/io/airbyte/integrations/destination/bigquery/write/typing_deduping/legacy_raw_tables/BigqueryTypingDedupingDatabaseInitialStatusGatherer.kt b/airbyte-integrations/connectors/destination-bigquery/src/main/kotlin/io/airbyte/integrations/destination/bigquery/write/typing_deduping/legacy_raw_tables/BigqueryTypingDedupingDatabaseInitialStatusGatherer.kt index 2b6756447f76..a83e46c83780 100644 --- a/airbyte-integrations/connectors/destination-bigquery/src/main/kotlin/io/airbyte/integrations/destination/bigquery/write/typing_deduping/legacy_raw_tables/BigqueryTypingDedupingDatabaseInitialStatusGatherer.kt +++ b/airbyte-integrations/connectors/destination-bigquery/src/main/kotlin/io/airbyte/integrations/destination/bigquery/write/typing_deduping/legacy_raw_tables/BigqueryTypingDedupingDatabaseInitialStatusGatherer.kt @@ -8,7 +8,6 @@ import com.google.cloud.bigquery.BigQuery import com.google.cloud.bigquery.QueryJobConfiguration import com.google.cloud.bigquery.TableId import io.airbyte.cdk.load.command.DestinationStream -import io.airbyte.integrations.destination.bigquery.BigQueryUtils import io.airbyte.cdk.load.orchestration.db.DatabaseInitialStatusGatherer import io.airbyte.cdk.load.orchestration.db.TableName import io.airbyte.cdk.load.orchestration.db.TableNames @@ -16,6 +15,7 @@ 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 { diff --git a/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/typing_deduping/direct_load_tables/BigqueryDirectLoadDatabaseInitialStatusGathererTest.kt b/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/typing_deduping/direct_load_tables/BigqueryDirectLoadDatabaseInitialStatusGathererTest.kt index 57957fd0a20f..2f04031f85bc 100644 --- a/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/typing_deduping/direct_load_tables/BigqueryDirectLoadDatabaseInitialStatusGathererTest.kt +++ b/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/typing_deduping/direct_load_tables/BigqueryDirectLoadDatabaseInitialStatusGathererTest.kt @@ -52,24 +52,25 @@ class BigqueryDirectLoadDatabaseInitialStatusGathererTest { 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) + fun `InterruptedException during getTable is wrapped as TransientErrorException`() = + runBlocking { + val interruptedException = InterruptedException("thread interrupted") + val bigQueryException = BigQueryException(0, "interrupted", interruptedException) - every { bigquery.getTable(any()) } throws bigQueryException - every { tempTableNameGenerator.generate(any()) } returns tempTableName + every { bigquery.getTable(any()) } 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()))) - ) + val tableNames = TableNames(finalTableName = tableName, rawTableName = null) + val catalog = + TableCatalog( + mapOf(stream to TableNameInfo(tableNames, ColumnNameMapping(emptyMap()))) + ) - assertThrows { gatherer.gatherInitialStatus(catalog) } + assertThrows { gatherer.gatherInitialStatus(catalog) } - // Clear interrupted status - Thread.interrupted() - } + // Clear interrupted status + Thread.interrupted() + } @Test fun `successful getTable returns status normally`() = runBlocking { @@ -81,9 +82,7 @@ class BigqueryDirectLoadDatabaseInitialStatusGathererTest { val tableNames = TableNames(finalTableName = tableName, rawTableName = null) val catalog = - TableCatalog( - mapOf(stream to TableNameInfo(tableNames, ColumnNameMapping(emptyMap()))) - ) + TableCatalog(mapOf(stream to TableNameInfo(tableNames, ColumnNameMapping(emptyMap())))) val result = gatherer.gatherInitialStatus(catalog) assertNotNull(result[stream]) @@ -98,9 +97,7 @@ class BigqueryDirectLoadDatabaseInitialStatusGathererTest { val tableNames = TableNames(finalTableName = tableName, rawTableName = null) val catalog = - TableCatalog( - mapOf(stream to TableNameInfo(tableNames, ColumnNameMapping(emptyMap()))) - ) + TableCatalog(mapOf(stream to TableNameInfo(tableNames, ColumnNameMapping(emptyMap())))) assertThrows { gatherer.gatherInitialStatus(catalog) } } diff --git a/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/typing_deduping/legacy_raw_tables/BigqueryTypingDedupingDatabaseInitialStatusGathererTest.kt b/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/typing_deduping/legacy_raw_tables/BigqueryTypingDedupingDatabaseInitialStatusGathererTest.kt index 6a685c1e5ac6..c404902bae84 100644 --- a/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/typing_deduping/legacy_raw_tables/BigqueryTypingDedupingDatabaseInitialStatusGathererTest.kt +++ b/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/typing_deduping/legacy_raw_tables/BigqueryTypingDedupingDatabaseInitialStatusGathererTest.kt @@ -48,23 +48,25 @@ class BigqueryTypingDedupingDatabaseInitialStatusGathererTest { private val finalTableName = TableName("test_namespace", "test_stream") @Test - fun `InterruptedException during getTable is wrapped as TransientErrorException`() = runBlocking { - val interruptedException = InterruptedException("thread interrupted") - val bigQueryException = BigQueryException(0, "interrupted", interruptedException) + fun `InterruptedException during getTable is wrapped as TransientErrorException`() = + runBlocking { + val interruptedException = InterruptedException("thread interrupted") + val bigQueryException = BigQueryException(0, "interrupted", interruptedException) - every { bq.getTable(any()) } throws bigQueryException + every { bq.getTable(any()) } throws bigQueryException - val tableNames = TableNames(rawTableName = rawTableName, finalTableName = finalTableName) - val catalog = - TableCatalog( - mapOf(stream to TableNameInfo(tableNames, ColumnNameMapping(emptyMap()))) - ) + val tableNames = + TableNames(rawTableName = rawTableName, finalTableName = finalTableName) + val catalog = + TableCatalog( + mapOf(stream to TableNameInfo(tableNames, ColumnNameMapping(emptyMap()))) + ) - assertThrows { gatherer.gatherInitialStatus(catalog) } + assertThrows { gatherer.gatherInitialStatus(catalog) } - // Clear interrupted status - Thread.interrupted() - } + // Clear interrupted status + Thread.interrupted() + } @Test fun `table not found returns status with null raw table state`() = runBlocking { @@ -72,9 +74,7 @@ class BigqueryTypingDedupingDatabaseInitialStatusGathererTest { val tableNames = TableNames(rawTableName = rawTableName, finalTableName = finalTableName) val catalog = - TableCatalog( - mapOf(stream to TableNameInfo(tableNames, ColumnNameMapping(emptyMap()))) - ) + TableCatalog(mapOf(stream to TableNameInfo(tableNames, ColumnNameMapping(emptyMap())))) val result = gatherer.gatherInitialStatus(catalog) assertNotNull(result[stream]) @@ -89,9 +89,7 @@ class BigqueryTypingDedupingDatabaseInitialStatusGathererTest { val tableNames = TableNames(rawTableName = rawTableName, finalTableName = finalTableName) val catalog = - TableCatalog( - mapOf(stream to TableNameInfo(tableNames, ColumnNameMapping(emptyMap()))) - ) + TableCatalog(mapOf(stream to TableNameInfo(tableNames, ColumnNameMapping(emptyMap())))) assertThrows { gatherer.gatherInitialStatus(catalog) } } From 437c48cfbca8f84f9234926ef4df1937c145989f Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 8 Jun 2026 11:59:23 +0000 Subject: [PATCH 3/4] fix: use io.airbyte.cdk.TransientErrorException from bulk CDK instead of io.airbyte.commons Co-Authored-By: bot_apk --- .../airbyte/integrations/destination/bigquery/BigQueryUtils.kt | 3 +-- .../bigquery/BigQueryUtilsExecuteBigQueryOperationTest.kt | 2 +- .../BigqueryDirectLoadDatabaseInitialStatusGathererTest.kt | 2 +- .../BigqueryTypingDedupingDatabaseInitialStatusGathererTest.kt | 2 +- 4 files changed, 4 insertions(+), 5 deletions(-) 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 3e6dbc664592..5efcf6b66123 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,8 +10,8 @@ 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 io.airbyte.commons.exceptions.TransientErrorException import java.util.* import java.util.stream.Collectors import org.apache.commons.lang3.StringUtils @@ -193,7 +193,6 @@ object BigQueryUtils { throw TransientErrorException( "BigQuery API call interrupted.", e, - "BigQuery operation interrupted, likely due to sync cancellation: ${e.message}", ) } throw e 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 9e2a6c75f6ac..00c6beb46a19 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 @@ -5,7 +5,7 @@ package io.airbyte.integrations.destination.bigquery import com.google.cloud.bigquery.BigQueryException -import io.airbyte.commons.exceptions.TransientErrorException +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 diff --git a/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/typing_deduping/direct_load_tables/BigqueryDirectLoadDatabaseInitialStatusGathererTest.kt b/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/typing_deduping/direct_load_tables/BigqueryDirectLoadDatabaseInitialStatusGathererTest.kt index 2f04031f85bc..7cdb44beff2d 100644 --- a/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/typing_deduping/direct_load_tables/BigqueryDirectLoadDatabaseInitialStatusGathererTest.kt +++ b/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/typing_deduping/direct_load_tables/BigqueryDirectLoadDatabaseInitialStatusGathererTest.kt @@ -8,6 +8,7 @@ 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 @@ -18,7 +19,6 @@ 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.commons.exceptions.TransientErrorException import io.airbyte.integrations.destination.bigquery.write.typing_deduping.direct_load_tables.BigqueryDirectLoadDatabaseInitialStatusGatherer import io.mockk.every import io.mockk.mockk diff --git a/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/typing_deduping/legacy_raw_tables/BigqueryTypingDedupingDatabaseInitialStatusGathererTest.kt b/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/typing_deduping/legacy_raw_tables/BigqueryTypingDedupingDatabaseInitialStatusGathererTest.kt index c404902bae84..84bda108a25f 100644 --- a/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/typing_deduping/legacy_raw_tables/BigqueryTypingDedupingDatabaseInitialStatusGathererTest.kt +++ b/airbyte-integrations/connectors/destination-bigquery/src/test/kotlin/io/airbyte/integrations/destination/bigquery/typing_deduping/legacy_raw_tables/BigqueryTypingDedupingDatabaseInitialStatusGathererTest.kt @@ -7,6 +7,7 @@ package io.airbyte.integrations.destination.bigquery.typing_deduping.legacy_raw_ import com.google.cloud.bigquery.BigQuery import com.google.cloud.bigquery.BigQueryException 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 @@ -16,7 +17,6 @@ import io.airbyte.cdk.load.orchestration.db.TableName import io.airbyte.cdk.load.orchestration.db.TableNames 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.commons.exceptions.TransientErrorException import io.airbyte.integrations.destination.bigquery.write.typing_deduping.legacy_raw_tables.BigqueryTypingDedupingDatabaseInitialStatusGatherer import io.mockk.every import io.mockk.mockk From 7ab9a63788c5797086bc6dca2fbf5b990193c679 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Mon, 8 Jun 2026 12:16:59 +0000 Subject: [PATCH 4/4] chore: update changelog with 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 d3f07603e6ca..fe681252ea34 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.20 | 2026-06-08 | [TBD](https://github.com/airbytehq/airbyte/pull/TBD) | Improved error handling for InterruptedException during sync cancellation. | +| 3.0.20 | 2026-06-08 | [79168](https://github.com/airbytehq/airbyte/pull/79168) | Improved error handling for InterruptedException during sync cancellation. | | 3.0.19 | 2026-05-21 | [78239](https://github.com/airbytehq/airbyte/pull/78239) | Promoting release candidate 3.0.19-rc.1 to a main version. | | 3.0.19-rc.1 | 2026-05-19 | [78239](https://github.com/airbytehq/airbyte/pull/78239) | Upgrade CDK to 1.0.13. Progressive rollout. | | 3.0.18 | 2026-03-31 | [75913](https://github.com/airbytehq/airbyte/pull/75913) | Finalize upgrade BigQuery Cloud dependencies and CDK version |