diff --git a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadChunkCallable.java b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadChunkCallable.java index 53724df8ac7a..ac55662cf24c 100644 --- a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadChunkCallable.java +++ b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadChunkCallable.java @@ -34,7 +34,10 @@ import com.google.api.gax.resumable.ChunkUploadRequest; import com.google.api.gax.resumable.ChunkUploadResponse; import com.google.api.gax.resumable.ResumableUploadStatus; +import com.google.api.gax.resumable.ResumableUploadStatusCode; import com.google.api.gax.rpc.ApiCallContext; +import com.google.api.gax.rpc.ApiException; +import com.google.api.gax.rpc.ApiExceptionFactory; import com.google.api.gax.rpc.ClientContext; import com.google.api.gax.rpc.UnaryCallable; import com.google.api.pathtemplate.PathTemplate; @@ -199,6 +202,9 @@ public void onClose(int statusCode, HttpJsonMetadata trailers) { response = responseParser.parse(stream); } future.set(ChunkUploadResponse.create(uploadStatus, response)); + } else if (uploadStatus == ResumableUploadStatus.FINAL) { + future.setException( + createServerRejectionException(statusCode, trailers.getException(), uploadStatus)); } else { Throwable cause = trailers.getException(); future.setException( @@ -211,5 +217,18 @@ public void onClose(int statusCode, HttpJsonMetadata trailers) { future.setException(t); } } + + private static ApiException createServerRejectionException( + int statusCode, @Nullable Throwable cause, ResumableUploadStatus uploadStatus) { + String message = "Server terminated upload session with HTTP status: " + statusCode; + if (cause != null && cause.getMessage() != null) { + message = cause.getMessage(); + } + return ApiExceptionFactory.createException( + message, + cause, + ResumableUploadStatusCode.of(HttpJsonStatusCode.of(statusCode), uploadStatus), + false); + } } } diff --git a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadQueryStatusCallable.java b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadQueryStatusCallable.java index 69d8042fd8b4..336ec1b79e03 100644 --- a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadQueryStatusCallable.java +++ b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadQueryStatusCallable.java @@ -34,7 +34,9 @@ import com.google.api.gax.resumable.QueryStatusRequest; import com.google.api.gax.resumable.QueryStatusResponse; import com.google.api.gax.resumable.ResumableUploadStatus; +import com.google.api.gax.resumable.ResumableUploadStatusCode; import com.google.api.gax.rpc.ApiCallContext; +import com.google.api.gax.rpc.ApiException; import com.google.api.gax.rpc.ApiExceptionFactory; import com.google.api.gax.rpc.ClientContext; import com.google.api.gax.rpc.StatusCode; @@ -246,6 +248,9 @@ public void onClose(int statusCode, HttpJsonMetadata trailers) { HttpJsonStatusCode.of(StatusCode.Code.INTERNAL), /* retryable= */ false)); } + } else if (uploadStatus == ResumableUploadStatus.FINAL) { + future.setException( + createServerRejectionException(statusCode, trailers.getException(), uploadStatus)); } else { Throwable cause = trailers.getException(); future.setException( @@ -260,5 +265,18 @@ public void onClose(int statusCode, HttpJsonMetadata trailers) { future.setException(t); } } + + private static ApiException createServerRejectionException( + int statusCode, @Nullable Throwable cause, ResumableUploadStatus uploadStatus) { + String message = "Server terminated upload session with HTTP status: " + statusCode; + if (cause != null && cause.getMessage() != null) { + message = cause.getMessage(); + } + return ApiExceptionFactory.createException( + message, + cause, + ResumableUploadStatusCode.of(HttpJsonStatusCode.of(statusCode), uploadStatus), + false); + } } } diff --git a/sdk-platform-java/gax-java/gax-httpjson/src/test/java/com/google/api/gax/httpjson/HttpJsonResumableUploadClientTest.java b/sdk-platform-java/gax-java/gax-httpjson/src/test/java/com/google/api/gax/httpjson/HttpJsonResumableUploadClientTest.java index c94513975dfd..9408dac5cc3f 100644 --- a/sdk-platform-java/gax-java/gax-httpjson/src/test/java/com/google/api/gax/httpjson/HttpJsonResumableUploadClientTest.java +++ b/sdk-platform-java/gax-java/gax-httpjson/src/test/java/com/google/api/gax/httpjson/HttpJsonResumableUploadClientTest.java @@ -46,12 +46,15 @@ import com.google.api.gax.resumable.QueryStatusResponse; import com.google.api.gax.resumable.ResumableUploadSession; import com.google.api.gax.resumable.ResumableUploadStatus; +import com.google.api.gax.resumable.ResumableUploadStatusCode; import com.google.api.gax.rpc.AbortedException; import com.google.api.gax.rpc.ApiCallContext; import com.google.api.gax.rpc.ClientContext; import com.google.api.gax.rpc.InternalException; +import com.google.api.gax.rpc.InvalidArgumentException; import com.google.api.gax.rpc.NotFoundException; import com.google.api.gax.rpc.StatusCode; +import com.google.api.gax.rpc.UnavailableException; import com.google.api.pathtemplate.PathTemplate; import com.google.common.base.Strings; import java.io.IOException; @@ -355,6 +358,61 @@ void uploadChunk_withCustomExtraHeaders_preservesHeaders() { .containsExactly("CustomChunkValue"); } + @Test + void uploadChunk_serverReturnsFinalStatusOnNon200_marksExceptionNonRetryable() { + MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse(); + httpResponse.setStatusCode(503); + httpResponse.addHeader("X-Goog-Upload-Status", "final"); + + CapturingHttpTransport transport = new CapturingHttpTransport(httpResponse); + HttpJsonResumableUploadClient client = createClient(transport); + + ChunkUploadRequest request = + ChunkUploadRequest.newBuilder() + .setUploadUrl(TEST_UPLOAD_URL) + .setPayload("data".getBytes(StandardCharsets.UTF_8)) + .setOffset(0L) + .setFinal(true) + .build(); + + UnavailableException ex = + assertThrows(UnavailableException.class, () -> client.uploadChunkCallable().call(request)); + assertThat(ex.isRetryable()).isFalse(); + assertThat(ex.getStatusCode().getCode()).isEqualTo(StatusCode.Code.UNAVAILABLE); + assertThat(ex.getStatusCode().getTransportCode()).isEqualTo(503); + assertThat(ex.getStatusCode()).isInstanceOf(ResumableUploadStatusCode.class); + assertThat(((ResumableUploadStatusCode) ex.getStatusCode()).getUploadStatus()) + .isEqualTo(ResumableUploadStatus.FINAL); + } + + @Test + void uploadChunk_serverReturnsFinalStatusOn400_throwsInvalidArgumentException() { + MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse(); + httpResponse.setStatusCode(400); + httpResponse.addHeader("X-Goog-Upload-Status", "final"); + + CapturingHttpTransport transport = new CapturingHttpTransport(httpResponse); + HttpJsonResumableUploadClient client = createClient(transport); + + ChunkUploadRequest request = + ChunkUploadRequest.newBuilder() + .setUploadUrl(TEST_UPLOAD_URL) + .setPayload("data".getBytes(StandardCharsets.UTF_8)) + .setOffset(0L) + .setFinal(false) + .build(); + + InvalidArgumentException ex = + assertThrows( + InvalidArgumentException.class, () -> client.uploadChunkCallable().call(request)); + assertThat(ex.isRetryable()).isFalse(); + assertThat(ex.getStatusCode().getCode()).isEqualTo(StatusCode.Code.INVALID_ARGUMENT); + assertThat(ex.getStatusCode().getTransportCode()).isEqualTo(400); + assertThat(ex.getStatusCode()).isInstanceOf(ResumableUploadStatusCode.class); + assertThat(((ResumableUploadStatusCode) ex.getStatusCode()).getUploadStatus()) + .isEqualTo(ResumableUploadStatus.FINAL); + } + @Test void uploadChunk_serverReturnsConflictOrError_throwsException() { MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse(); diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadStatusCode.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadStatusCode.java new file mode 100644 index 000000000000..9748b48f5815 --- /dev/null +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadStatusCode.java @@ -0,0 +1,88 @@ +/* + * Copyright 2026 Google LLC + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of Google LLC nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + */ +package com.google.api.gax.resumable; + +import static com.google.common.base.Preconditions.checkNotNull; + +import com.google.api.core.BetaApi; +import com.google.api.core.InternalApi; +import com.google.api.gax.rpc.StatusCode; +import java.util.Objects; +import org.jspecify.annotations.NullMarked; +import org.jspecify.annotations.Nullable; + +@NullMarked +@BetaApi +@InternalApi +public final class ResumableUploadStatusCode implements StatusCode { + private final StatusCode delegate; + private final ResumableUploadStatus uploadStatus; + + private ResumableUploadStatusCode(StatusCode delegate, ResumableUploadStatus uploadStatus) { + this.delegate = checkNotNull(delegate, "delegate must not be null"); + this.uploadStatus = checkNotNull(uploadStatus, "uploadStatus must not be null"); + } + + public static ResumableUploadStatusCode of( + StatusCode delegate, ResumableUploadStatus uploadStatus) { + return new ResumableUploadStatusCode(delegate, uploadStatus); + } + + @Override + public Code getCode() { + return delegate.getCode(); + } + + @Override + public @Nullable Object getTransportCode() { + return delegate.getTransportCode(); + } + + public ResumableUploadStatus getUploadStatus() { + return uploadStatus; + } + + @Override + public boolean equals(@Nullable Object o) { + if (this == o) { + return true; + } + if (!(o instanceof ResumableUploadStatusCode)) { + return false; + } + ResumableUploadStatusCode that = (ResumableUploadStatusCode) o; + return Objects.equals(delegate, that.delegate) && uploadStatus == that.uploadStatus; + } + + @Override + public int hashCode() { + return Objects.hash(delegate, uploadStatus); + } +} diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadErrorClassifier.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadErrorClassifier.java index b531c56ab72a..aa03cb0fe076 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadErrorClassifier.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadErrorClassifier.java @@ -29,6 +29,8 @@ */ package com.google.api.gax.rpc; +import com.google.api.gax.resumable.ResumableUploadStatus; +import com.google.api.gax.resumable.ResumableUploadStatusCode; import com.google.common.collect.ImmutableMap; import java.net.SocketTimeoutException; import java.util.Objects; @@ -86,6 +88,13 @@ static Category classify(Throwable t, ResumableUploadCommand command) { ApiException apiException = (ApiException) t; StatusCode statusCode = apiException.getStatusCode(); + if (statusCode instanceof ResumableUploadStatusCode) { + ResumableUploadStatusCode uploadStatusCode = (ResumableUploadStatusCode) statusCode; + if (uploadStatusCode.getUploadStatus() == ResumableUploadStatus.FINAL) { + return Category.FATAL; + } + } + // HttpJsonApiExceptionFactory wraps low-level network timeouts as UNKNOWN. if (statusCode.getCode() == StatusCode.Code.UNKNOWN) { if (apiException.getCause() instanceof SocketTimeoutException) { 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 f11fd34c0721..ffb38b68929f 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 @@ -250,9 +250,53 @@ private void fail(Throwable t) { if (inFlight != null) { inFlight.cancel(true); } + Throwable augmented = augmentWithUrl(t); progressTracker.onFailed(); - closePayload(); - resultFuture.setException(t); + 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 baseMessage = firstNonNull(t.getMessage(), t.getClass().getSimpleName()); + boolean alreadyContainsUrl = baseMessage.contains(url); + if (alreadyContainsUrl && !(t instanceof IllegalStateException)) { + return t; + } + String augmentedMessage = baseMessage; + if (!alreadyContainsUrl) { + 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 FailedPreconditionException( + augmentedMessage, t, FAILED_PRECONDITION_STATUS_CODE, false); + } 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() { @@ -332,6 +376,19 @@ 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; 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 57d794a0fca1..e6f26c0fe2a9 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 @@ -97,7 +97,8 @@ void realignTo(long committedOffset) throws IOException { 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", + + " 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 6cd64105fa12..5ab6d7b11898 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 @@ -53,6 +53,7 @@ import com.google.api.gax.resumable.ResumableUploadProgress; import com.google.api.gax.resumable.ResumableUploadSession; import com.google.api.gax.resumable.ResumableUploadStatus; +import com.google.api.gax.resumable.ResumableUploadStatusCode; import com.google.api.gax.rpc.testing.FakeCallContext; import java.io.ByteArrayInputStream; import java.io.IOException; @@ -109,7 +110,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); } @@ -332,7 +337,7 @@ void testUploadCallable_chunkFailure_failsFuture() { callable.futureCall("resource-path", streamOf("data"), null); ExecutionException exception = assertThrows(ExecutionException.class, future::get); - assertThat(exception.getCause()).isInstanceOf(IllegalStateException.class); + assertThat(exception.getCause()).isInstanceOf(FailedPreconditionException.class); assertThat(exception.getCause()).hasMessageThat().contains("chunk error"); } @@ -621,7 +626,7 @@ void testRecovery_queryNullOffset_failsFatal() throws Exception { callable.futureCall("resource-path", streamOf("hello"), null); ExecutionException exception = assertThrows(ExecutionException.class, future::get); - assertThat(exception.getCause()).isInstanceOf(IllegalStateException.class); + assertThat(exception.getCause()).isInstanceOf(FailedPreconditionException.class); assertThat(exception.getCause()) .hasMessageThat() .contains("did not include a committed offset"); @@ -681,7 +686,7 @@ void testRecovery_offsetBelowBufferBase_failsFatal() throws Exception { callable.futureCall("resource-path", streamOf("0123456789abcdef"), null); ExecutionException exception = assertThrows(ExecutionException.class, future::get); - assertThat(exception.getCause()).isInstanceOf(IllegalStateException.class); + assertThat(exception.getCause()).isInstanceOf(FailedPreconditionException.class); assertThat(exception.getCause()).hasMessageThat().contains("below buffer base offset"); } @@ -758,7 +763,7 @@ void testRecovery_queryMissingStatusHeader_failsFatal() throws Exception { callable.futureCall("resource-path", streamOf("hello"), null); ExecutionException exception = assertThrows(ExecutionException.class, future::get); - assertThat(exception.getCause()).isInstanceOf(IllegalStateException.class); + assertThat(exception.getCause()).isInstanceOf(FailedPreconditionException.class); assertThat(exception.getCause()) .hasMessageThat() .contains("missing X-Goog-Upload-Status header"); @@ -1304,6 +1309,193 @@ 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(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 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"); + } + + @Test + void testRecovery_serverRejectionWithFinalStatus_failsFatalWithoutRetryOrRecovery() { + String sessionUrl = "https://upload.url/server-rejection-test"; + stubStartSession(sessionUrl); + StatusCode rejectionStatusCode = + ResumableUploadStatusCode.of( + new HttpStatusStatusCode(400, StatusCode.Code.INVALID_ARGUMENT), + ResumableUploadStatus.FINAL); + ApiException rejectionException = + ApiExceptionFactory.createException("Invalid chunk", null, rejectionStatusCode, false); + when(mockChunkCallable.futureCall(any(ChunkUploadRequest.class), any())) + .thenReturn(ApiFutures.immediateFailedFuture(rejectionException)); + + ResumableUploadFuture future = + callable.futureCall("resource-path", streamOf("hello"), null); + + ExecutionException ex = assertThrows(ExecutionException.class, future::get); + 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); + } + 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/ResumableUploadErrorClassifierTest.java b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadErrorClassifierTest.java index 31a5cbb460e6..2244de3f44a6 100644 --- a/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadErrorClassifierTest.java +++ b/sdk-platform-java/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadErrorClassifierTest.java @@ -40,6 +40,8 @@ import static com.google.common.truth.Truth.assertThat; import static org.junit.jupiter.api.Assertions.assertThrows; +import com.google.api.gax.resumable.ResumableUploadStatus; +import com.google.api.gax.resumable.ResumableUploadStatusCode; import com.google.api.gax.rpc.StatusCode.Code; import java.io.IOException; import java.net.SocketException; @@ -75,6 +77,27 @@ private static ApiException createApiException( "HTTP " + httpStatus, cause, statusCode(httpStatus, code), false); } + @Test + void testServerRejectionStatusCode_isFatalEvenWithRetryableOrRecoverableHttpCode() { + ApiException final503 = + ApiExceptionFactory.createException( + "HTTP 503", + null, + ResumableUploadStatusCode.of( + statusCode(503, Code.UNAVAILABLE), ResumableUploadStatus.FINAL), + false); + assertThat(ResumableUploadErrorClassifier.classify(final503, UPLOAD)).isEqualTo(FATAL); + + ApiException final400 = + ApiExceptionFactory.createException( + "HTTP 400", + null, + ResumableUploadStatusCode.of( + statusCode(400, Code.INVALID_ARGUMENT), ResumableUploadStatus.FINAL), + false); + assertThat(ResumableUploadErrorClassifier.classify(final400, UPLOAD)).isEqualTo(FATAL); + } + @Test void testTransientHttpErrors_areTransientAcrossCommands() { ApiException error503 = createApiException(503, Code.UNAVAILABLE); 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 40351740863c..95b586e57f11 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 @@ -152,6 +152,7 @@ void testRealignToBelowBaseOffset_throwsIllegalStateException() throws IOExcepti assertThat(exception.getMessage()).contains("4"); assertThat(exception.getMessage()).contains("8"); assertThat(exception.getMessage()).contains(UPLOAD_URL); + assertThat(exception.getMessage()).contains("seekable stream"); } @Test