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.

Author: damccorm

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWriteUnshardedRecords.java

@@ -792,9 +792,17 @@ long flush(
 
               // 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

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/bigquery/StorageApiWritesShardedRecords.java

@@ -1066,7 +1066,25 @@ public void process(
 
         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) {

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/testing/FakeDatasetService.java

@@ -238,6 +238,16 @@ void commit() {
 
   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 @@ Exceptions.StorageException tryInitialize() throws Exception {
       @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();

sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/bigquery/BigQueryIOWriteTest.java

@@ -1659,6 +1659,69 @@ public void testStreamingStorageApiWriteWithErrorHandling() throws Exception {
     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);