From 9beb28492fd67d39bdcd54134929df5c96136cf7 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. --- .../gax/rpc/ResumableUploadFutureImpl.java | 45 ++++- .../api/gax/rpc/RewindableStreamBuffer.java | 3 +- .../rpc/ResumableUploadCallableImplTest.java | 167 +++++++++++++++++- .../gax/rpc/RewindableStreamBufferTest.java | 1 + 4 files changed, 211 insertions(+), 5 deletions(-) diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadFutureImpl.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadFutureImpl.java index 87fb4dba0912..dddc6ee185c1 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadFutureImpl.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadFutureImpl.java @@ -241,9 +241,48 @@ private void fail(Throwable t) { 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() { 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 996ca227093c..0dfae0f0e700 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 @@ -102,7 +102,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)); } 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 0719c6ced5e2..f819427bca84 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 @@ -110,7 +110,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); } @@ -1343,6 +1347,167 @@ 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(ResumableUploadStatus.ACTIVE, null))) + .thenReturn( + ApiFutures.immediateFailedFuture( + createApiException(400, StatusCode.Code.INVALID_ARGUMENT))); + + when(mockQueryCallable.futureCall(any(QueryStatusRequest.class), any())) + .thenReturn( + ApiFutures.immediateFuture( + createQueryResponse(false, 4L, null, ResumableUploadStatus.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); + } + + @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 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; 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 a8a2f69d84cf..c95ee2e1b644 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 @@ -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