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..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,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,27 @@ 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, + ) + } + 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..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 @@ -15,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 { @@ -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..00c6beb46a19 --- /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.cdk.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..7cdb44beff2d --- /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,104 @@ +/* + * 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.TransientErrorException +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.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..84bda108a25f --- /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,96 @@ +/* + * 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.TransientErrorException +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.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..fe681252ea34 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 | [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 |