Improve BigQueryIO Storage Write API error messages

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);