From 087add0192b49aaccd5f7ee0cd0cdd48d6d81a26 Mon Sep 17 00:00:00 2001 From: whowes Date: Sat, 12 Sep 2026 17:29:35 +0000 Subject: [PATCH] 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. --- .../rpc/ResumableUploadChunkCoordinator.java | 41 +++++- .../api/gax/rpc/RewindableStreamBuffer.java | 3 +- .../rpc/ResumableUploadCallableImplTest.java | 134 +++++++++++++++++- .../gax/rpc/RewindableStreamBufferTest.java | 1 + 4 files changed, 174 insertions(+), 5 deletions(-) diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadChunkCoordinator.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadChunkCoordinator.java index 9a77ee962856..f8b66ef9254b 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadChunkCoordinator.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadChunkCoordinator.java @@ -274,14 +274,49 @@ private void finish(@Nullable ResponseT response, @Nullable Throwable error) { progressTracker.onFinalized(totalBytes); result.set(response); } else { + Throwable augmented = augmentWithUrl(error); if (closeError != null) { - error.addSuppressed(closeError); + augmented.addSuppressed(closeError); } - progressTracker.onFailed(error, uploadSessionUrl); - result.setException(error); + progressTracker.onFailed(augmented, uploadSessionUrl); + result.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 augmentedMessage = + (message != null ? message : t.getClass().getSimpleName()) + " (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 @Nullable IOException closePayload() { synchronized (lock) { if (payloadClosed) { diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/RewindableStreamBuffer.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/RewindableStreamBuffer.java index 24a27dafabcf..73ce3aef0a79 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/RewindableStreamBuffer.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/RewindableStreamBuffer.java @@ -101,7 +101,8 @@ void realignTo(long committedOffset) throws IOException { throw UploadErrors.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)); } diff --git a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadCallableImplTest.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadCallableImplTest.java index 8d637ae81c61..d295dd91dbdb 100644 --- a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadCallableImplTest.java +++ b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadCallableImplTest.java @@ -109,7 +109,11 @@ void setUp() { callContext = FakeCallContext.createDefault(); executor = Executors.newScheduledThreadPool(2); clientContext = - ClientContext.newBuilder().setDefaultCallContext(callContext).setExecutor(executor).build(); + ClientContext.newBuilder() + .setDefaultCallContext(callContext) + .setExecutor(executor) + .setEndpoint("https://test.endpoint.com") + .build(); callable = new ResumableUploadCallableImpl<>(mockClient, defaultSettings, clientContext); } @@ -1347,6 +1351,134 @@ 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 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 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 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 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> hungChunk = SettableApiFuture.create(); + when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any())).thenReturn(hungChunk); + + ResumableUploadCallSettings settings = + defaultSettings.toBuilder().setGlobalTimeout(Duration.ofMillis(50)).build(); + + ResumableUploadFuture future = + callable.futureCall("resource-path", streamOf("hello"), null, settings); + + ExecutionException ex = assertThrows(ExecutionException.class, future::get); + 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(false, null))) + .thenReturn( + ApiFutures.immediateFailedFuture( + createApiException(400, StatusCode.Code.INVALID_ARGUMENT))); + + when(mockQueryCallable.futureCall(any(QueryStatusRequest.class), any())) + .thenReturn(ApiFutures.immediateFuture(createQueryResponse(false, 4L, null, "active"))); + + byte[] data = new byte[16]; + ResumableUploadFuture 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); + } + private static class HttpStatusStatusCode implements StatusCode { private final int httpStatus; private final StatusCode.Code code; diff --git a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/RewindableStreamBufferTest.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/RewindableStreamBufferTest.java index 7b3ac1f7ce8a..df3bc3f25b4c 100644 --- a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/RewindableStreamBufferTest.java +++ b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/RewindableStreamBufferTest.java @@ -154,6 +154,7 @@ void testRealignToBelowBaseOffset_throwsFailedPreconditionException_classifiedFa assertThat(exception.getMessage()).contains("4"); assertThat(exception.getMessage()).contains("8"); assertThat(exception.getMessage()).contains(UPLOAD_URL); + assertThat(exception.getMessage()).contains("seekable stream"); // Must be classified as FATAL by ResumableUploadErrorClassifier ResumableUploadErrorClassifier.Category category =