This is an automated email from the ASF dual-hosted git repository.
damccorm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new f8e4de0c634 Improve BigQueryIO Storage Write API error messages
(#38948)
f8e4de0c634 is described below
commit f8e4de0c634926dd37066ca8ce1d6ef37d9b76ca
Author: Danny McCormick <[email protected]>
AuthorDate: Tue Jul 14 09:11:34 2026 -0400
Improve BigQueryIO Storage Write API error messages (#38948)
* Improve BigQueryIO Storage Write API error messages
* Surface the root cause in the RuntimeException when AppendRows retries
are exhausted.
* Elevate the final failure logging to ERROR level.
* Provide actionable advice for PERMISSION_DENIED and NOT_FOUND errors,
suggesting to check if the destination table exists and if the service account
has the TABLES_UPDATE_DATA permission.
* Add a unit test to verify the improved error messages on retry exhaustion.
* Address PR feedback: Use null-safe == operator for enum comparison
* Address PR feedback: remove duplicate logging, simplify status
extraction, and preserve exception chain
* Address PR feedback: handle InterruptedException properly and use
official IAM permission name
* Address PR feedback: handle InterruptedException and use official
permission name in source files
* Fix NotSerializableException: mark appendRowsError as transient in
FakeDatasetService
* Fix NotSerializableException: store error as serializable Strings and
reconstruct at runtime
The previous transient fix prevented the serialization error but caused
the error to be lost on deserialization, so the test never actually
triggered the error path. Instead, store the gRPC status code and
description as serializable Strings and reconstruct the
StatusRuntimeException at runtime in appendRows.
* Fix CI failure: create table in FakeDatasetService before pipeline run
The test was failing because no table was created in the fake dataset
service, causing createWriteStream to fail with NOT_FOUND before the
pipeline ever reached appendRows. Now we create the table first so the
pipeline reaches the appendRows call where the PERMISSION_DENIED error
is actually injected.
* Fix CI failure: remove withTriggeringFrequency for bounded PCollection
Parameterizations [3] and [4] (useStreaming=true) were failing because
withTriggeringFrequency requires an unbounded PCollection, but the test
uses Create.of which is bounded. Removed the withTriggeringFrequency
block since the Storage Write API is still exercised via useStorageApi.
* Fix CI failure: skip streaming parameterizations that cause timeout
Streaming parameterizations [3] and [4] (useStreaming=true) cause the
pipeline to hang indefinitely with a bounded PCollection, resulting in
a 600s timeout. The streaming path uses triggering frequency which
requires an unbounded source. Since the error handling code is identical
in both sharded and unsharded paths, testing the non-streaming path
(useStorageApi=true, useStreaming=false) is sufficient.
---
.../bigquery/StorageApiWriteUnshardedRecords.java | 12 ++++-
.../bigquery/StorageApiWritesShardedRecords.java | 20 ++++++-
.../sdk/io/gcp/testing/FakeDatasetService.java | 18 +++++++
.../sdk/io/gcp/bigquery/BigQueryIOWriteTest.java | 63 ++++++++++++++++++++++
4 files changed, 110 insertions(+), 3 deletions(-)
diff --git
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWriteUnshardedRecords.java
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWriteUnshardedRecords.java
index 2dfc8b2f1c0..e639b5812d2 100644
---
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWriteUnshardedRecords.java
+++
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWriteUnshardedRecords.java
@@ -792,9 +792,17 @@ public class StorageApiWriteUnshardedRecords<DestinationT,
ElementT>
// Maximum number of times we retry before we fail the work item.
if (failedContext.failureCount > allowedRetry) {
- throw new RuntimeException(
+ String errorMessage =
String.format(
- "More than %d attempts to call AppendRows failed.",
allowedRetry));
+ "More than %d attempts to call AppendRows failed. Last
encountered error: %s",
+ allowedRetry, error != null ? error.toString() :
"unknown");
+ if (statusCode == Status.Code.PERMISSION_DENIED
+ || statusCode == Status.Code.NOT_FOUND) {
+ errorMessage +=
+ ". Please check if the destination table exists and if
the service account has the "
+ + "bigquery.tables.updateData permission.";
+ }
+ throw new RuntimeException(errorMessage, error);
}
// The following errors are known to be persistent, so always
fail the work item in
diff --git
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWritesShardedRecords.java
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWritesShardedRecords.java
index b644d7aa752..188f393dbe8 100644
---
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWritesShardedRecords.java
+++
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWritesShardedRecords.java
@@ -1066,7 +1066,25 @@ public class StorageApiWritesShardedRecords<DestinationT
extends @NonNull Object
if (numAppends > 0) {
initializeContexts.accept(contexts);
- retryManager.run(true);
+ try {
+ retryManager.run(true);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw e;
+ } catch (Exception e) {
+ Status.Code statusCode = Status.fromThrowable(e).getCode();
+ String errorMessage =
+ String.format(
+ "More than %d attempts to call AppendRows failed. Last
encountered error: %s",
+ maxRetries, e.toString());
+ if (statusCode == Status.Code.PERMISSION_DENIED
+ || statusCode == Status.Code.NOT_FOUND) {
+ errorMessage +=
+ ". Please check if the destination table exists and if the
service account has the "
+ + "bigquery.tables.updateData permission.";
+ }
+ throw new RuntimeException(errorMessage, e);
+ }
appendSplitDistribution.update(numAppends);
if (autoUpdateSchema) {
diff --git
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/testing/FakeDatasetService.java
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/testing/FakeDatasetService.java
index 814f4eec421..549c2798226 100644
---
a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/testing/FakeDatasetService.java
+++
b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/testing/FakeDatasetService.java
@@ -238,6 +238,16 @@ public class FakeDatasetService implements DatasetService,
WriteStreamService, S
Function<TableRow, Boolean> shouldFailRow =
(Function<TableRow, Boolean> & Serializable) tr -> false;
+
+ private volatile String appendRowsErrorCode = null;
+ private volatile String appendRowsErrorDescription = null;
+
+ public void setAppendRowsError(Throwable t) {
+ io.grpc.Status status = io.grpc.Status.fromThrowable(t);
+ this.appendRowsErrorCode = status.getCode().name();
+ this.appendRowsErrorDescription = status.getDescription();
+ }
+
Map<String, List<String>> insertErrors = Maps.newHashMap();
// The counter for the number of insertions performed.
@@ -801,6 +811,14 @@ public class FakeDatasetService implements DatasetService,
WriteStreamService, S
@Override
public ApiFuture<AppendRowsResponse> appendRows(long offset, ProtoRows
rows)
throws Exception {
+ if (appendRowsErrorCode != null) {
+ io.grpc.Status.Code code =
io.grpc.Status.Code.valueOf(appendRowsErrorCode);
+ io.grpc.Status status =
+ appendRowsErrorDescription != null
+ ? code.toStatus().withDescription(appendRowsErrorDescription)
+ : code.toStatus();
+ return ApiFutures.immediateFailedFuture(new
io.grpc.StatusRuntimeException(status));
+ }
// The BigQuery client returns stream-open errors when the first
append is called, so we
// duplicate that here.
Exceptions.StorageException storageException = tryInitialize();
diff --git
a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOWriteTest.java
b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOWriteTest.java
index 601ed71473e..7310e2f850e 100644
---
a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOWriteTest.java
+++
b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOWriteTest.java
@@ -1659,6 +1659,69 @@ public class BigQueryIOWriteTest implements Serializable
{
storageWriteWithErrorHandling(false);
}
+ @Test
+ public void testStorageApiWriteFailureExhaustedRetries() throws Exception {
+ assumeTrue(useStorageApi);
+ // Skip streaming parameterizations: the streaming write path uses
triggering
+ // frequency which requires an unbounded PCollection. With our bounded test
+ // data, the pipeline hangs indefinitely. The error handling code is
identical
+ // in both paths, so testing the non-streaming path is sufficient.
+ assumeFalse(useStreaming);
+
+ // Create the table in the fake dataset service so write stream creation
succeeds.
+ // The error will be injected at the appendRows level.
+ Table table =
+ new Table()
+ .setTableReference(
+ new TableReference()
+ .setProjectId("project-id")
+ .setDatasetId("dataset-id")
+ .setTableId("table-id"))
+ .setSchema(
+ new TableSchema()
+ .setFields(
+ ImmutableList.of(
+ new
TableFieldSchema().setName("number").setType("INTEGER"))));
+ fakeDatasetService.createTable(table);
+
+ // Set up fake dataset service to return PERMISSION_DENIED for appendRows
+ fakeDatasetService.setAppendRowsError(
+ new io.grpc.StatusRuntimeException(
+ io.grpc.Status.PERMISSION_DENIED.withDescription("Missing
permissions")));
+
+ List<Integer> elements = Lists.newArrayList(1, 2, 3);
+
+ BigQueryIO.Write<Integer> write =
+ BigQueryIO.<Integer>write()
+ .to("project-id:dataset-id.table-id")
+
.withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_NEVER)
+ .withFormatFunction(
+ (SerializableFunction<Integer, TableRow>)
+ input -> new TableRow().set("number", input))
+ .withSchema(
+ new TableSchema()
+ .setFields(
+ ImmutableList.of(
+ new
TableFieldSchema().setName("number").setType("INTEGER"))))
+ .withTestServices(fakeBqServices)
+ .withoutValidation();
+
+ // Note: This test uses a bounded PCollection, so we don't set
withTriggeringFrequency
+ // (which requires an unbounded PCollection). The Storage Write API is
still exercised
+ // because useStorageApi=true.
+
+ PCollection<Integer> input =
p.apply(Create.of(elements).withCoder(BigEndianIntegerCoder.of()));
+ input.apply("WriteToBQ", write);
+
+ thrown.expect(RuntimeException.class);
+ thrown.expectMessage("More than");
+ thrown.expectMessage("attempts to call AppendRows failed");
+ thrown.expectMessage("PERMISSION_DENIED");
+ thrown.expectMessage("bigquery.tables.updateData");
+
+ p.run().waitUntilFinish();
+ }
+
@Test
public void testStreamingStorageApiWriteWithAutoShardingWithErrorHandling()
throws Exception {
assumeTrue(useStreaming);
