feat(gax): standardize mapping terminal resumable upload exceptions to ApiException

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() {
@@ -150,8 +150,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);
@@ -263,8 +262,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);
@@ -274,8 +272,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);
@@ -290,8 +287,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/ResumableUploadFutureImpl.java

@@ -212,14 +212,15 @@
   }
 
   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";
     }
-    fail(ApiExceptionFactory.createException(message, null, TIMEOUT_STATUS_CODE, false));
+    fail(
+        ApiExceptionFactory.createException(
+            message, null, statusCodeOf(StatusCode.Code.DEADLINE_EXCEEDED), false));
   }
 
   private void succeed(@Nullable ResponseT result) {
@@ -229,7 +230,7 @@
       }
       inFlightFuture = null;
     }
-    closePayload();
+    closePayload(null);
     resultFuture.set(result);
   }
 
@@ -243,19 +244,44 @@
       inFlightFuture = null;
     }
     inFlight.cancel(true);
+    ApiException terminal = toTerminalException(t);
     progressTracker.onFailed();
-    closePayload();
-    resultFuture.setException(t);
+    closePayload(terminal);
+    resultFuture.setException(terminal);
   }
 
-  private void closePayload() {
+  private ApiException toTerminalException(Throwable t) {
+    if (t instanceof ApiException && uploadSessionUrl == null) {
+      return (ApiException) t;
+    }
+    String message = firstNonNull(t.getMessage(), t.getClass().getSimpleName());
+    if (uploadSessionUrl != null) {
+      message = message + " (upload URL: " + uploadSessionUrl + ")";
+    }
+    // Rethrow with URL included in message.
+    if (t instanceof ApiException) {
+      ApiException apiException = (ApiException) t;
+      return ApiExceptionFactory.createException(
+          message,
+          apiException.getCause(),
+          apiException.getStatusCode(),
+          apiException.isRetryable(),
+          apiException.getErrorDetails());
+    }
+    return ApiExceptionFactory.createException(
+        message, t, statusCodeOf(StatusCode.Code.UNKNOWN), false);
+  }
+
+  private void closePayload(@Nullable Throwable primaryException) {
     if (payload == null) {
       return;
     }
     try {
       payload.close();
-    } catch (IOException ignored) {
-      // Suppressed during stream cleanup
+    } catch (Throwable e) {
+      if (primaryException != null) {
+        primaryException.addSuppressed(e);
+      }
     }
   }
 
@@ -283,7 +309,7 @@
     }
     inFlight.cancel(mayInterruptIfRunning);
     progressTracker.onFailed();
-    closePayload();
+    closePayload(null);
     return cancelled;
   }
 
@@ -308,16 +334,17 @@
     return resultFuture.get(timeout, unit);
   }
 
-  private static final StatusCode TIMEOUT_STATUS_CODE =
-      new StatusCode() {
-        @Override
-        public StatusCode.Code getCode() {
-          return StatusCode.Code.DEADLINE_EXCEEDED;
-        }
+  private static StatusCode statusCodeOf(StatusCode.Code code) {
+    return new StatusCode() {
+      @Override
+      public StatusCode.Code getCode() {
+        return code;
+      }
 
-        @Override
-        public @Nullable Object getTransportCode() {
-          return null;
-        }
-      };
+      @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);

sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadCallableImplTest.java

@@ -58,6 +58,7 @@
 import com.google.api.gax.rpc.testing.FakeCallContext;
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
+import java.io.InputStream;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.List;
@@ -305,7 +306,7 @@ void testUploadCallable_startFailure_failsFuture() {
         callable.futureCall("resource-path", payloadOf("data"), null);
 
     ExecutionException exception = assertThrows(ExecutionException.class, future::get);
-    assertThat(exception.getCause()).isInstanceOf(IllegalStateException.class);
+    assertThat(exception.getCause()).isInstanceOf(UnknownException.class);
     assertThat(exception.getCause()).hasMessageThat().contains("start failed");
     verifyNoInteractions(mockChunkCallable);
   }
@@ -320,7 +321,7 @@ void testUploadCallable_chunkFailure_failsFuture() {
         callable.futureCall("resource-path", payloadOf("data"), null);
 
     ExecutionException exception = assertThrows(ExecutionException.class, future::get);
-    assertThat(exception.getCause()).isInstanceOf(IllegalStateException.class);
+    assertThat(exception.getCause()).isInstanceOf(UnknownException.class);
     assertThat(exception.getCause()).hasMessageThat().contains("chunk error");
   }
 
@@ -609,7 +610,7 @@ void testRecovery_queryNullOffset_failsFatal() throws Exception {
         callable.futureCall("resource-path", payloadOf("hello"), null);
 
     ExecutionException exception = assertThrows(ExecutionException.class, future::get);
-    assertThat(exception.getCause()).isInstanceOf(IllegalStateException.class);
+    assertThat(exception.getCause()).isInstanceOf(UnknownException.class);
     assertThat(exception.getCause())
         .hasMessageThat()
         .contains("did not include a committed offset");
@@ -669,7 +670,7 @@ void testRecovery_offsetBelowBufferBase_failsFatal() throws Exception {
         callable.futureCall("resource-path", payloadOf("0123456789abcdef"), null);
 
     ExecutionException exception = assertThrows(ExecutionException.class, future::get);
-    assertThat(exception.getCause()).isInstanceOf(IllegalStateException.class);
+    assertThat(exception.getCause()).isInstanceOf(UnknownException.class);
     assertThat(exception.getCause()).hasMessageThat().contains("below buffer base offset");
   }
 
@@ -746,7 +747,7 @@ void testRecovery_queryMissingStatusHeader_failsFatal() throws Exception {
         callable.futureCall("resource-path", payloadOf("hello"), null);
 
     ExecutionException exception = assertThrows(ExecutionException.class, future::get);
-    assertThat(exception.getCause()).isInstanceOf(IllegalStateException.class);
+    assertThat(exception.getCause()).isInstanceOf(UnknownException.class);
     assertThat(exception.getCause())
         .hasMessageThat()
         .contains("missing X-Goog-Upload-Status header");
@@ -824,6 +825,7 @@ void testRecovery_serverRejectionWithFinalStatus_failsFatalWithoutRetryOrRecover
     assertThat(ex.getCause()).isInstanceOf(InvalidArgumentException.class);
     ApiException cause = (ApiException) ex.getCause();
     assertThat(cause.getStatusCode().getTransportCode()).isEqualTo(400);
+    assertThat(cause.getMessage()).contains(sessionUrl);
     verify(mockChunkCallable, times(1)).futureCall(any(), any());
     verifyNoInteractions(mockQueryCallable);
   }
@@ -1259,6 +1261,77 @@ List<ResumableUploadProgress> awaitTerminal() throws InterruptedException {
     }
   }
 
+  @Test
+  void testActionableErrors_startFailure_preservesOriginalException() {
+    ApiException startError = createApiException(401, StatusCode.Code.UNAUTHENTICATED);
+    when(mockStartCallable.futureCall(any(), any()))
+        .thenReturn(ApiFutures.immediateFailedFuture(startError));
+
+    ResumableUploadFuture<String> future =
+        callable.futureCall("resource-path", payloadOf("hello"), null);
+
+    ExecutionException ex = assertThrows(ExecutionException.class, future::get);
+    assertThat(ex.getCause()).isSameInstanceAs(startError);
+    assertThat(future.getUploadSessionUrl()).isNull();
+  }
+
+  @Test
+  void testActionableErrors_preservesErrorDetailsAndCauseChain() {
+    String sessionUrl = "https://upload.url/chunk-error-details-test";
+    stubStartSession(sessionUrl);
+    ErrorDetails errorDetails = ErrorDetails.builder().build();
+    IOException transportCause = new IOException("403 Forbidden");
+    ApiException original =
+        ApiExceptionFactory.createException(
+            "HTTP 403",
+            transportCause,
+            new HttpStatusStatusCode(403, StatusCode.Code.PERMISSION_DENIED),
+            false,
+            errorDetails);
+    when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
+        .thenReturn(ApiFutures.immediateFailedFuture(original));
+
+    ResumableUploadFuture<String> future =
+        callable.futureCall("resource-path", payloadOf("hello"), null);
+
+    ExecutionException ex = assertThrows(ExecutionException.class, future::get);
+    assertThat(ex.getCause()).isInstanceOf(ApiException.class);
+    ApiException cause = (ApiException) ex.getCause();
+    assertThat(cause.getMessage()).contains(sessionUrl);
+    assertThat(cause.getCause()).isSameInstanceAs(transportCause);
+    assertThat(cause.getErrorDetails()).isSameInstanceAs(errorDetails);
+    assertThat(future.getUploadSessionUrl()).isEqualTo(sessionUrl);
+  }
+
+  @Test
+  void testUploadCallable_failureOutcome_attachesCloseExceptionViaAddSuppressed() {
+    when(mockStartCallable.futureCall(any(), any()))
+        .thenReturn(ApiFutures.immediateFailedFuture(new IllegalStateException("upload failed")));
+
+    InputStream failingStream =
+        new InputStream() {
+          @Override
+          public int read() {
+            return -1;
+          }
+
+          @Override
+          public void close() throws IOException {
+            throw new IOException("stream close error");
+          }
+        };
+
+    ResumableUploadFuture<String> future =
+        callable.futureCall("resource-path", () -> failingStream, null);
+    ExecutionException exception = assertThrows(ExecutionException.class, future::get);
+    assertThat(exception.getCause()).isInstanceOf(UnknownException.class);
+    assertThat(exception.getCause().getSuppressed()).asList().hasSize(1);
+    assertThat(exception.getCause().getSuppressed()[0]).isInstanceOf(IOException.class);
+    assertThat(exception.getCause().getSuppressed()[0])
+        .hasMessageThat()
+        .contains("stream close error");
+  }
+
   private static class HttpStatusStatusCode implements StatusCode {
     private final int httpStatus;
     private final StatusCode.Code code;