feat(gax): surface actionable error messages with upload session URL and stream requirements

Augments terminal failure exceptions with the active upload session URL to aid debugging and session recovery. Clarifies error messages when a server committed offset falls below the buffer base offset.

Author: whowes

java-showcase/gapic-showcase/src/test/java/com/google/showcase/v1beta1/it/ITResumableUpload.java

@@ -277,6 +277,7 @@ void testChunkFatalError(@TempDir Path tempDir) throws Exception {
       assertThat(exception.getCause()).isInstanceOf(NotFoundException.class);
       NotFoundException notFoundException = (NotFoundException) exception.getCause();
       assertThat(notFoundException.getStatusCode().getCode()).isEqualTo(StatusCode.Code.NOT_FOUND);
+      assertThat(notFoundException.getMessage()).contains(future.getUploadSessionUrl());
     }
   }
 

sdk-platform-java/gax-java/gax-httpjson/src/test/java/com/google/api/gax/httpjson/HttpJsonResumableUploadClientTest.java

@@ -566,7 +566,9 @@ void queryStatus_serverReturnsFinalStatusOnNon200_marksExceptionNonRetryable() {
     UnavailableException ex =
         assertThrows(UnavailableException.class, () -> client.queryStatusCallable().call(request));
     assertThat(ex.isRetryable()).isFalse();
-    assertThat(ex).hasMessageThat().contains("Server terminated upload session with HTTP status: 503");
+    assertThat(ex)
+        .hasMessageThat()
+        .contains("Server terminated upload session with HTTP status: 503");
     assertThat(ex.getStatusCode()).isInstanceOf(ResumableUploadStatusCode.class);
     assertThat(((ResumableUploadStatusCode) ex.getStatusCode()).getUploadStatus())
         .isEqualTo(ResumableUploadStatus.FINAL);

sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadStatusCode.java

@@ -39,8 +39,8 @@
 import org.jspecify.annotations.Nullable;
 
 /**
- * A {@link StatusCode} that preserves the transport status code of a failed resumable upload command
- * and carries the {@code X-Goog-Upload-Status} header value from the response.
+ * A {@link StatusCode} that preserves the transport status code of a failed resumable upload
+ * command and carries the {@code X-Goog-Upload-Status} header value from the response.
  *
  * <p>Used to propagate a server rejection (a non-2xx response with {@code X-Goog-Upload-Status:
  * final}) from the transport layer to the resumable upload retry logic, so that the rejection is

sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadChunkCoordinator.java

@@ -99,7 +99,7 @@ final class ResumableUploadChunkCoordinator<ResponseT> {
     this.executor = checkNotNull(executor, "executor must not be null");
     this.recoverySettings = recoveryAlgorithm.createFirstAttempt();
     this.progressTracker = checkNotNull(progressTracker, "progressTracker must not be null");
-    this.buffer = new RewindableStreamBuffer(payload, chunkSize, uploadUrl);
+    this.buffer = new RewindableStreamBuffer(payload, chunkSize);
   }
 
   ApiFuture<ResponseT> getFuture() {
@@ -149,8 +149,7 @@ public void onSuccess(ChunkUploadResponse<ResponseT> response) {
               if (response.getUploadStatus() == ResumableUploadStatus.UNKNOWN) {
                 recover(
                     new IllegalStateException(
-                        "Chunk upload response missing X-Goog-Upload-Status header for upload URL: "
-                            + uploadUrl));
+                        "Chunk upload response missing X-Goog-Upload-Status header"));
               } else {
                 try {
                   handleChunkResponse(response);
@@ -262,8 +261,7 @@ private boolean tryRegisterInFlightFuture(Future<?> future) {
   private void handleQueryResponse(QueryStatusResponse<ResponseT> queryResponse)
       throws IOException {
     if (queryResponse.getUploadStatus() == ResumableUploadStatus.UNKNOWN) {
-      throw new IllegalStateException(
-          "Query status response missing X-Goog-Upload-Status header for upload URL: " + uploadUrl);
+      throw new IllegalStateException("Query status response missing X-Goog-Upload-Status header");
     }
     if (queryResponse.getUploadStatus() == ResumableUploadStatus.FINAL) {
       progressTracker.onFinalized(buffer.getBufferBaseOffset() + buffer.getPayload().length);
@@ -273,8 +271,7 @@ private void handleQueryResponse(QueryStatusResponse<ResponseT> queryResponse)
     Long committedOffset = queryResponse.getCommittedOffset();
     if (committedOffset == null) {
       throw new IllegalStateException(
-          "Incomplete query status response did not include a committed offset for upload URL: "
-              + uploadUrl);
+          "Incomplete query status response did not include a committed offset");
     }
     progressTracker.onOffsetReceived(committedOffset);
     buffer.realignTo(committedOffset);
@@ -289,8 +286,7 @@ private void handleChunkResponse(ChunkUploadResponse<ResponseT> response) throws
       uploadResultFuture.setException(
           new IllegalStateException(
               "Upload stream ended and final chunk was transmitted, but server returned"
-                  + " incomplete status for upload URL: "
-                  + uploadUrl));
+                  + " incomplete status"));
     } else {
       madeProgressSinceRecovery = true;
       buffer.fill();

sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadFuture.java

@@ -41,6 +41,12 @@
  * <p>The payload {@link java.io.InputStream} supplied when initiating the upload is managed by this
  * future and will be closed automatically upon completion, failure, or cancellation.
  *
+ * <p>If the upload fails, {@link #get()} throws an {@link java.util.concurrent.ExecutionException}
+ * whose cause is an {@link ApiException}. If the upload is cancelled, {@link #get()} throws a
+ * {@link java.util.concurrent.CancellationException}. When the upload session URL is known, it is
+ * included in the exception message. A rejection by the server is terminal and is not retried or
+ * recovered.
+ *
  * @param <ResponseT> the type of the final response message returned once the upload completes
  */
 @BetaApi

sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadFutureImpl.java

@@ -210,10 +210,9 @@ public void onFailure(Throwable t) {
   }
 
   private void onTimeout() {
-    String sessionUrl = uploadSessionUrl;
     String message;
-    if (sessionUrl != null) {
-      message = "Resumable upload timed out for session: " + sessionUrl;
+    if (uploadSessionUrl != null) {
+      message = "Resumable upload timed out";
     } else {
       message = "Resumable upload timed out before session initiation completed";
     }
@@ -243,9 +242,45 @@ private void fail(Throwable t) {
     if (inFlight != null) {
       inFlight.cancel(true);
     }
+    Throwable terminal = toTerminalException(t);
     progressTracker.onFailed();
-    closePayload();
-    resultFuture.setException(t);
+    try {
+      payload.close();
+    } catch (Throwable closeException) {
+      terminal.addSuppressed(closeException);
+    }
+    resultFuture.setException(terminal);
+  }
+
+  private Throwable toTerminalException(Throwable t) {
+    String url = uploadSessionUrl;
+    if (t instanceof ApiException && url == null) {
+      return t;
+    }
+    String message = firstNonNull(t.getMessage(), t.getClass().getSimpleName());
+    if (url != null) {
+      message = message + " (upload URL: " + url + ")";
+    }
+    ApiException terminal;
+    if (t instanceof ApiException) {
+      ApiException apiException = (ApiException) t;
+      terminal =
+          ApiExceptionFactory.createException(
+              message,
+              apiException,
+              apiException.getStatusCode(),
+              apiException.isRetryable(),
+              apiException.getErrorDetails());
+    } else if (t instanceof IllegalStateException) {
+      terminal =
+          new FailedPreconditionException(message, t, FAILED_PRECONDITION_STATUS_CODE, false);
+    } else {
+      terminal = ApiExceptionFactory.createException(message, t, UNKNOWN_STATUS_CODE, false);
+    }
+    for (Throwable suppressed : t.getSuppressed()) {
+      terminal.addSuppressed(suppressed);
+    }
+    return terminal;
   }
 
   private void closePayload() {
@@ -324,6 +359,32 @@ public StatusCode.Code getCode() {
           return StatusCode.Code.DEADLINE_EXCEEDED;
         }
 
+        @Override
+        public @Nullable Object getTransportCode() {
+          return null;
+        }
+      };
+
+  private static final StatusCode FAILED_PRECONDITION_STATUS_CODE =
+      new StatusCode() {
+        @Override
+        public StatusCode.Code getCode() {
+          return StatusCode.Code.FAILED_PRECONDITION;
+        }
+
+        @Override
+        public @Nullable Object getTransportCode() {
+          return null;
+        }
+      };
+
+  private static final StatusCode UNKNOWN_STATUS_CODE =
+      new StatusCode() {
+        @Override
+        public StatusCode.Code getCode() {
+          return StatusCode.Code.UNKNOWN;
+        }
+
         @Override
         public @Nullable Object getTransportCode() {
           return null;

sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/RewindableStreamBuffer.java

@@ -46,19 +46,17 @@ final class RewindableStreamBuffer {
 
   private final InputStream inputStream;
   private final int chunkSize;
-  private final String uploadUrl;
   private final byte[] buffer;
 
   private long bufferBaseOffset;
   private int payloadLength;
   private boolean isFinal;
   private boolean streamExhausted;
 
-  RewindableStreamBuffer(InputStream inputStream, int chunkSize, String uploadUrl) {
+  RewindableStreamBuffer(InputStream inputStream, int chunkSize) {
     this.inputStream = checkNotNull(inputStream, "inputStream must not be null");
     checkArgument(chunkSize > 0, "chunkSize must be > 0");
     this.chunkSize = chunkSize;
-    this.uploadUrl = checkNotNull(uploadUrl, "uploadUrl must not be null");
     this.buffer = new byte[chunkSize];
     this.bufferBaseOffset = 0L;
     this.payloadLength = 0;
@@ -95,17 +93,16 @@ void realignTo(long committedOffset) throws IOException {
     if (committedOffset < bufferBaseOffset) {
       throw new IllegalStateException(
           String.format(
-              "Server committed offset %d is below buffer base offset %d for upload URL %s; cannot"
-                  + " rewind stream before buffer base",
-              committedOffset, bufferBaseOffset, uploadUrl));
+              "Server committed offset %d is below buffer base offset %d, which the server already"
+                  + " acknowledged; the upload cannot continue and must be restarted.",
+              committedOffset, bufferBaseOffset));
     }
 
     if (committedOffset > bufferBaseOffset + payloadLength) {
       throw new IllegalStateException(
           String.format(
-              "Server committed offset %d is beyond current buffer window [%d, %d] for upload URL"
-                  + " %s",
-              committedOffset, bufferBaseOffset, bufferBaseOffset + payloadLength, uploadUrl));
+              "Server committed offset %d is beyond current buffer window [%d, %d]",
+              committedOffset, bufferBaseOffset, bufferBaseOffset + payloadLength));
     }
 
     int committedWithinBuffer = (int) (committedOffset - bufferBaseOffset);