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

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

@@ -256,9 +256,48 @@
     if (inFlight != null) {
       inFlight.cancel(true);
     }
-    progressTracker.onFailed(t, uploadSessionUrl);
-    closePayload();
-    resultFuture.setException(t);
+    Throwable augmented = augmentWithUrl(t);
+    progressTracker.onFailed(augmented, uploadSessionUrl);
+    try {
+      payload.close();
+    } catch (Throwable closeException) {
+      augmented.addSuppressed(closeException);
+    }
+    resultFuture.setException(augmented);
+  }
+
+  private Throwable augmentWithUrl(Throwable t) {
+    String url = uploadSessionUrl;
+    if (url == null || url.isEmpty()) {
+      return t;
+    }
+    String message = t.getMessage();
+    if (message != null && message.contains(url)) {
+      return t;
+    }
+    String baseMessage = message != null ? message : t.getClass().getSimpleName();
+    String augmentedMessage = baseMessage + " (upload URL: " + url + ")";
+    Throwable augmented = t;
+    if (t instanceof ApiException) {
+      ApiException apiException = (ApiException) t;
+      augmented =
+          ApiExceptionFactory.createException(
+              augmentedMessage,
+              apiException,
+              apiException.getStatusCode(),
+              apiException.isRetryable(),
+              apiException.getErrorDetails());
+    } else if (t instanceof IllegalStateException) {
+      augmented = new IllegalStateException(augmentedMessage, t);
+    } else if (t instanceof IOException) {
+      augmented = new IOException(augmentedMessage, t);
+    }
+    if (augmented != t) {
+      for (Throwable suppressed : t.getSuppressed()) {
+        augmented.addSuppressed(suppressed);
+      }
+    }
+    return augmented;
   }
 
   private void closePayload() {

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

@@ -95,7 +95,8 @@ void realignTo(long committedOffset) throws IOException {
       throw protocolViolation(
           String.format(
               "Server committed offset %d is below buffer base offset %d for upload URL %s; cannot"
-                  + " rewind stream before buffer base",
+                  + " rewind stream before buffer base. A seekable stream is required to rewind to"
+                  + " earlier offsets.",
               committedOffset, bufferBaseOffset, uploadUrl));
     }
 

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

@@ -107,7 +107,11 @@ void setUp() {
 
     defaultSettings = ResumableUploadCallSettings.newBuilder().setChunkSize(8).build();
     callContext = FakeCallContext.createDefault();
-    clientContext = ClientContext.newBuilder().setDefaultCallContext(callContext).build();
+    clientContext =
+        ClientContext.newBuilder()
+            .setDefaultCallContext(callContext)
+            .setEndpoint("https://test.endpoint.com")
+            .build();
     executor = Executors.newSingleThreadExecutor();
     callable = new ResumableUploadCallableImpl<>(mockClient, defaultSettings, clientContext);
   }
@@ -1254,6 +1258,168 @@ void testProgressListener_orderingUnderConcurrency_pinsSequentialExecutor() thro
     }
   }
 
+  @Test
+  void testActionableErrors_startFailure_preservesOriginalExceptionWithoutEndpointSuffix() {
+    ApiException startError = createApiException(401, StatusCode.Code.UNAUTHENTICATED);
+    when(mockStartCallable.futureCall(any(), any()))
+        .thenReturn(ApiFutures.immediateFailedFuture(startError));
+
+    ResumableUploadFuture<String> future =
+        callable.futureCall("resource-path", streamOf("hello"), null);
+
+    ExecutionException ex = assertThrows(ExecutionException.class, future::get);
+    assertThat(ex.getCause()).isSameInstanceAs(startError);
+    assertThat(ex.getCause().getMessage()).doesNotContain("endpoint:");
+    assertThat(future.getUploadSessionUrl()).isNull();
+  }
+
+  @Test
+  void testActionableErrors_chunkFailure_messageContainsUploadSessionUrl() {
+    String sessionUrl = "https://upload.url/chunk-error-test";
+    stubStartSession(sessionUrl);
+    when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
+        .thenReturn(
+            ApiFutures.immediateFailedFuture(
+                createApiException(403, StatusCode.Code.PERMISSION_DENIED)));
+
+    ResumableUploadFuture<String> future =
+        callable.futureCall("resource-path", streamOf("hello"), null);
+
+    ExecutionException ex = assertThrows(ExecutionException.class, future::get);
+    assertThat(ex.getCause()).isInstanceOf(ApiException.class);
+    assertThat(ex.getCause().getMessage()).contains(sessionUrl);
+    assertThat(future.getUploadSessionUrl()).isEqualTo(sessionUrl);
+  }
+
+  @Test
+  void testActionableErrors_preservesErrorDetailsCauseChainAndSuppressedExceptions() {
+    String sessionUrl = "https://upload.url/chunk-error-details-test";
+    stubStartSession(sessionUrl);
+    ErrorDetails errorDetails = ErrorDetails.builder().build();
+    ApiException original =
+        ApiExceptionFactory.createException(
+            "HTTP 403",
+            null,
+            new HttpStatusStatusCode(403, StatusCode.Code.PERMISSION_DENIED),
+            false,
+            errorDetails);
+    IOException suppressed = new IOException("underlying stream error");
+    original.addSuppressed(suppressed);
+    when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
+        .thenReturn(ApiFutures.immediateFailedFuture(original));
+
+    ResumableUploadFuture<String> future =
+        callable.futureCall("resource-path", streamOf("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(original);
+    assertThat(cause.getErrorDetails()).isSameInstanceAs(errorDetails);
+    assertThat(cause.getSuppressed()).asList().contains(suppressed);
+    assertThat(future.getUploadSessionUrl()).isEqualTo(sessionUrl);
+  }
+
+  @Test
+  void testActionableErrors_recoveryFailure_messageContainsUploadSessionUrl() {
+    String sessionUrl = "https://upload.url/recovery-error-test";
+    stubStartSession(sessionUrl);
+    when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
+        .thenReturn(
+            ApiFutures.immediateFailedFuture(
+                createApiException(400, StatusCode.Code.INVALID_ARGUMENT)));
+    when(mockQueryCallable.futureCall(any(QueryStatusRequest.class), any()))
+        .thenReturn(
+            ApiFutures.immediateFailedFuture(
+                createApiException(403, StatusCode.Code.PERMISSION_DENIED)));
+
+    ResumableUploadFuture<String> future =
+        callable.futureCall("resource-path", streamOf("hello"), null);
+
+    ExecutionException ex = assertThrows(ExecutionException.class, future::get);
+    assertThat(ex.getCause()).isInstanceOf(ApiException.class);
+    assertThat(ex.getCause().getMessage()).contains(sessionUrl);
+    assertThat(future.getUploadSessionUrl()).isEqualTo(sessionUrl);
+  }
+
+  @Test
+  void testActionableErrors_globalTimeoutFailure_messageContainsUploadSessionUrl() {
+    String sessionUrl = "https://upload.url/timeout-error-test";
+    stubStartSession(sessionUrl);
+    SettableApiFuture<ChunkUploadResponse<String>> hungChunk = SettableApiFuture.create();
+    when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any())).thenReturn(hungChunk);
+
+    ResumableUploadCallSettings settings =
+        defaultSettings.toBuilder().setGlobalTimeout(Duration.ofMillis(50)).build();
+
+    ResumableUploadFuture<String> future =
+        callable.futureCall("resource-path", streamOf("hello"), null, settings);
+
+    ExecutionException ex =
+        assertThrows(ExecutionException.class, () -> future.get(5, TimeUnit.SECONDS));
+    assertThat(ex.getCause()).isInstanceOf(DeadlineExceededException.class);
+    assertThat(ex.getCause().getMessage()).contains(sessionUrl);
+    assertThat(future.getUploadSessionUrl()).isEqualTo(sessionUrl);
+  }
+
+  @Test
+  void testActionableErrors_rewindFailure_surfacesActionableSeekableStreamMessage() {
+    String sessionUrl = "https://upload.url/rewind-error-test";
+    stubStartSession(sessionUrl);
+    when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any()))
+        .thenReturn(
+            ApiFutures.immediateFuture(
+                ChunkUploadResponse.create(ResumableUploadStatus.ACTIVE, null)))
+        .thenReturn(
+            ApiFutures.immediateFailedFuture(
+                createApiException(400, StatusCode.Code.INVALID_ARGUMENT)));
+
+    when(mockQueryCallable.futureCall(any(QueryStatusRequest.class), any()))
+        .thenReturn(
+            ApiFutures.immediateFuture(
+                createQueryResponse(4L, null, ResumableUploadStatus.ACTIVE)));
+
+    byte[] data = new byte[16];
+    ResumableUploadFuture<String> future =
+        callable.futureCall("resource-path", new ByteArrayInputStream(data), null);
+
+    ExecutionException ex = assertThrows(ExecutionException.class, future::get);
+    assertThat(ex.getCause()).isInstanceOf(FailedPreconditionException.class);
+    assertThat(ex.getCause().getMessage()).contains(sessionUrl);
+    assertThat(ex.getCause().getMessage()).contains("seekable stream");
+    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(IllegalStateException.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;

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

@@ -150,6 +150,7 @@ void testRealignToBelowBaseOffset_throwsFailedPreconditionException() throws IOE
     assertThat(exception.getMessage()).contains("4");
     assertThat(exception.getMessage()).contains("8");
     assertThat(exception.getMessage()).contains(UPLOAD_URL);
+    assertThat(exception.getMessage()).contains("seekable stream");
   }
 
   @Test