From 4d366d898cfd63f4624fc60e106c9dd7525a21d3 Mon Sep 17 00:00:00 2001 From: Maksym Tymoshyk Date: Mon, 31 Aug 2026 16:55:22 +0300 Subject: [PATCH] Treat a 404 from BigQuery table and dataset deletes as success BigQueryIO deletes the temporary tables and datasets it creates. A delete can succeed at BigQuery and still have its work item fail to commit afterwards; the runner then replays the work item and the replayed delete gets a 404 because the first attempt already removed the resource. deleteTable and deleteDataset passed ALWAYS_RETRY, so that 404 was retried MAX_RPC_RETRIES times and then thrown. In a streaming job the work item retries forever, which stalls a drain. Both now use DONT_RETRY_NOT_FOUND and swallow an item-not-found error, matching how getTable already handles a 404. Every other status keeps its existing retry and failure behaviour. --- CHANGES.md | 1 + .../io/gcp/bigquery/BigQueryServicesImpl.java | 63 ++++++++++++++----- .../bigquery/BigQueryServicesImplTest.java | 40 ++++++++++++ 3 files changed, 87 insertions(+), 17 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index 045b95c4e76b..7cba471ac6db 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -93,6 +93,7 @@ * (Java) KafkaIO dynamic reads no longer require the obsolete `beam_fn_api` experiment ([#29998](https://github.com/apache/beam/issues/29998)). * (Prism) Self-checkpointing splittable DoFns now resume after their requested delay instead of immediately, so polling SDFs no longer busy-spin ([#39848](https://github.com/apache/beam/issues/39848)). * (Java) MongoDbIO read splitting now preserves non-ObjectId `_id` types (e.g. string ids) instead of failing to parse the generated range filters ([#39900](https://github.com/apache/beam/issues/39900)). +* (Java) BigQueryIO now treats a 404 when deleting a temporary table or dataset as success, so a replayed work item whose earlier attempt already deleted it no longer retries forever ([#24997](https://github.com/apache/beam/issues/24997)). ## Security Fixes diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImpl.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImpl.java index 14765a65ff0b..cec50a45e00a 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImpl.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImpl.java @@ -805,20 +805,35 @@ Table tryCreateTable(Table table, BackOff backoff, Sleeper sleeper) throws IOExc * *

Tries executing the RPC for at most {@code MAX_RPC_RETRIES} times until it succeeds. * + *

A table that BigQuery reports as not found is treated as deleted successfully, since that + * is the state the caller asked for. + * * @throws IOException if it exceeds {@code MAX_RPC_RETRIES} attempts. */ @Override public void deleteTable(TableReference tableRef) throws IOException, InterruptedException { - executeWithRetries( - client - .tables() - .delete(tableRef.getProjectId(), tableRef.getDatasetId(), tableRef.getTableId()), - String.format( - "Unable to delete table: %s, aborting after %d retries.", - tableRef.getTableId(), MAX_RPC_RETRIES), - Sleeper.DEFAULT, - createDefaultBackoff(), - ALWAYS_RETRY); + try { + executeWithRetries( + client + .tables() + .delete(tableRef.getProjectId(), tableRef.getDatasetId(), tableRef.getTableId()), + String.format( + "Unable to delete table: %s, aborting after %d retries.", + tableRef.getTableId(), MAX_RPC_RETRIES), + Sleeper.DEFAULT, + createDefaultBackoff(), + DONT_RETRY_NOT_FOUND); + } catch (IOException e) { + if (!errorExtractor.itemNotFound(e)) { + throw e; + } + + // a delete can succeed at bigquery and still have its work item fail to commit afterwards. + // the runner then replays that work item, and the replayed delete gets a 404 because the + // first attempt already removed the table. failing here would make the work item retry + // forever, which in a streaming job stalls the drain indefinitely + LOG.info("Table {} is already deleted, treating as success.", tableRef.getTableId()); + } } @Override @@ -951,18 +966,32 @@ private void createDataset( * *

Tries executing the RPC for at most {@code MAX_RPC_RETRIES} times until it succeeds. * + *

A dataset that BigQuery reports as not found is treated as deleted successfully, since + * that is the state the caller asked for. + * * @throws IOException if it exceeds {@code MAX_RPC_RETRIES} attempts. */ @Override public void deleteDataset(String projectId, String datasetId) throws IOException, InterruptedException { - executeWithRetries( - client.datasets().delete(projectId, datasetId), - String.format( - "Unable to delete table: %s, aborting after %d retries.", datasetId, MAX_RPC_RETRIES), - Sleeper.DEFAULT, - createDefaultBackoff(), - ALWAYS_RETRY); + try { + executeWithRetries( + client.datasets().delete(projectId, datasetId), + String.format( + "Unable to delete table: %s, aborting after %d retries.", + datasetId, MAX_RPC_RETRIES), + Sleeper.DEFAULT, + createDefaultBackoff(), + DONT_RETRY_NOT_FOUND); + } catch (IOException e) { + if (!errorExtractor.itemNotFound(e)) { + throw e; + } + + // see deleteTable: a replayed work item can find the dataset its own earlier attempt + // already removed, and treating that 404 as a failure would retry forever + LOG.info("Dataset {} is already deleted, treating as success.", datasetId); + } } static class InsertBatchofRowsCallable implements Callable> { diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImplTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImplTest.java index 3902fb1fca33..9583e29491f4 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImplTest.java +++ b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryServicesImplTest.java @@ -569,6 +569,46 @@ public void testGetTableThrows() throws Exception { tableRef, Collections.emptyList(), null, BackOff.STOP_BACKOFF, Sleeper.DEFAULT); } + @Test + public void testDeleteTableNotFoundSucceeds() throws IOException, InterruptedException { + setupMockResponses( + response -> { + when(response.getContentType()).thenReturn(Json.MEDIA_TYPE); + when(response.getStatusCode()).thenReturn(404); + }); + + BigQueryServicesImpl.DatasetServiceImpl datasetService = + new BigQueryServicesImpl.DatasetServiceImpl(bigquery, PipelineOptionsFactory.create()); + + TableReference tableRef = + new TableReference() + .setProjectId("projectId") + .setDatasetId("datasetId") + .setTableId("tableId"); + + datasetService.deleteTable(tableRef); + + // exactly one response is prepared, so a retry of the 404 would trip the Verify inside the mock + // request. the assertion is therefore both "did not throw" and "did not retry" + verifyAllResponsesAreRead(); + } + + @Test + public void testDeleteDatasetNotFoundSucceeds() throws IOException, InterruptedException { + setupMockResponses( + response -> { + when(response.getContentType()).thenReturn(Json.MEDIA_TYPE); + when(response.getStatusCode()).thenReturn(404); + }); + + BigQueryServicesImpl.DatasetServiceImpl datasetService = + new BigQueryServicesImpl.DatasetServiceImpl(bigquery, PipelineOptionsFactory.create()); + + datasetService.deleteDataset("projectId", "datasetId"); + + verifyAllResponsesAreRead(); + } + @Test public void testIsTableEmptySucceeds() throws Exception { TableReference tableRef =