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 =