From 7c2dd99073a947e1c97a05929ce93ac849c6d87c Mon Sep 17 00:00:00 2001 From: whowes Date: Sat, 12 Sep 2026 17:29:35 +0000 Subject: [PATCH] feat(gax): standardize mapping terminal resumable upload exceptions to ApiException --- .../rpc/ResumableUploadChunkCoordinator.java | 14 ++-- .../gax/rpc/ResumableUploadFutureImpl.java | 71 +++++++++++----- .../api/gax/rpc/RewindableStreamBuffer.java | 15 ++-- .../rpc/ResumableUploadCallableImplTest.java | 83 +++++++++++++++++-- .../gax/rpc/RewindableStreamBufferTest.java | 30 +++---- 5 files changed, 148 insertions(+), 65 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 cfbd06d6a89b..6104e67ccc31 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 @@ -99,7 +99,7 @@ final class ResumableUploadChunkCoordinator { 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 getFuture() { @@ -150,8 +150,7 @@ public void onSuccess(ChunkUploadResponse 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 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 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 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(); 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 8bdb9cd9cfe3..d7a23f865a32 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 @@ -212,14 +212,15 @@ public void onFailure(Throwable t) { } 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 @@ private void succeed(@Nullable ResponseT result) { } inFlightFuture = null; } - closePayload(); + closePayload(null); resultFuture.set(result); } @@ -243,19 +244,44 @@ private void fail(Throwable t) { 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 @@ public boolean cancel(boolean mayInterruptIfRunning) { } inFlight.cancel(mayInterruptIfRunning); progressTracker.onFailed(); - closePayload(); + closePayload(null); return cancelled; } @@ -308,16 +334,17 @@ public ResponseT get(long timeout, TimeUnit unit) 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; + } + }; + } } 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 222014410440..0030f0282793 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 @@ -46,7 +46,6 @@ final class RewindableStreamBuffer { private final InputStream inputStream; private final int chunkSize; - private final String uploadUrl; private final byte[] buffer; private long bufferBaseOffset; @@ -54,11 +53,10 @@ final class RewindableStreamBuffer { 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); 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 c0560210a281..f03c94bb0e64 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 @@ -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 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 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 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 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; 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 fde9f12e7c10..6ca4b8020b54 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 @@ -41,13 +41,10 @@ class RewindableStreamBufferTest { - private static final String UPLOAD_URL = "https://upload.example.com/session-1"; - @Test void testExactMultiplePayloads() throws IOException { byte[] data = "0123456789abcdef".getBytes(StandardCharsets.UTF_8); // 16 bytes, chunk size 8 - RewindableStreamBuffer buffer = - new RewindableStreamBuffer(new ByteArrayInputStream(data), 8, UPLOAD_URL); + RewindableStreamBuffer buffer = new RewindableStreamBuffer(new ByteArrayInputStream(data), 8); // Chunk 0: 8 bytes buffer.fill(); @@ -74,8 +71,7 @@ void testExactMultiplePayloads() throws IOException { @Test void testShortFinalChunk() throws IOException { byte[] data = "short".getBytes(StandardCharsets.UTF_8); // 5 bytes, chunk size 8 - RewindableStreamBuffer buffer = - new RewindableStreamBuffer(new ByteArrayInputStream(data), 8, UPLOAD_URL); + RewindableStreamBuffer buffer = new RewindableStreamBuffer(new ByteArrayInputStream(data), 8); buffer.fill(); assertThat(buffer.getBufferBaseOffset()).isEqualTo(0L); @@ -87,7 +83,7 @@ void testShortFinalChunk() throws IOException { @Test void testZeroBytePayload() throws IOException { RewindableStreamBuffer buffer = - new RewindableStreamBuffer(new ByteArrayInputStream(new byte[0]), 8, UPLOAD_URL); + new RewindableStreamBuffer(new ByteArrayInputStream(new byte[0]), 8); buffer.fill(); assertThat(buffer.getBufferBaseOffset()).isEqualTo(0L); @@ -100,8 +96,7 @@ void testZeroBytePayload() throws IOException { void testRealignToMidBufferOffset_compactsAndTopsUp() throws IOException { // 20 bytes: chunk size 8 byte[] data = "0123456789ABCDEFGHIJ".getBytes(StandardCharsets.UTF_8); - RewindableStreamBuffer buffer = - new RewindableStreamBuffer(new ByteArrayInputStream(data), 8, UPLOAD_URL); + RewindableStreamBuffer buffer = new RewindableStreamBuffer(new ByteArrayInputStream(data), 8); // Initial fill: "01234567" (bytes 0..7) buffer.fill(); @@ -121,8 +116,7 @@ void testRealignToMidBufferOffset_compactsAndTopsUp() throws IOException { @Test void testRealignToBufferBaseOffset_isNoOp() throws IOException { byte[] data = "0123456789".getBytes(StandardCharsets.UTF_8); - RewindableStreamBuffer buffer = - new RewindableStreamBuffer(new ByteArrayInputStream(data), 8, UPLOAD_URL); + RewindableStreamBuffer buffer = new RewindableStreamBuffer(new ByteArrayInputStream(data), 8); buffer.fill(); assertThat(buffer.getPayload()).hasLength(8); @@ -138,8 +132,7 @@ void testRealignToBufferBaseOffset_isNoOp() throws IOException { @Test void testRealignToBelowBaseOffset_throwsIllegalStateException() throws IOException { byte[] data = "0123456789abcdef".getBytes(StandardCharsets.UTF_8); - RewindableStreamBuffer buffer = - new RewindableStreamBuffer(new ByteArrayInputStream(data), 8, UPLOAD_URL); + RewindableStreamBuffer buffer = new RewindableStreamBuffer(new ByteArrayInputStream(data), 8); // Advanced to chunk 1 (base offset 8) buffer.fill(); @@ -152,14 +145,13 @@ 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("must be restarted"); } @Test void testRealignToBeyondBufferWindow_throwsIllegalStateException() throws IOException { byte[] data = "0123456789abcdef".getBytes(StandardCharsets.UTF_8); - RewindableStreamBuffer buffer = - new RewindableStreamBuffer(new ByteArrayInputStream(data), 8, UPLOAD_URL); + RewindableStreamBuffer buffer = new RewindableStreamBuffer(new ByteArrayInputStream(data), 8); // Initial fill at 0: window is [0, 8] buffer.fill(); @@ -172,15 +164,13 @@ void testRealignToBeyondBufferWindow_throwsIllegalStateException() throws IOExce assertThat(exception.getMessage()).contains("10"); assertThat(exception.getMessage()).contains("8"); - assertThat(exception.getMessage()).contains(UPLOAD_URL); } @Test void testRealignToMidBufferOffset_reachingEofMarksFinal() throws IOException { // 10 bytes total, chunk size 8 byte[] data = "0123456789".getBytes(StandardCharsets.UTF_8); - RewindableStreamBuffer buffer = - new RewindableStreamBuffer(new ByteArrayInputStream(data), 8, UPLOAD_URL); + RewindableStreamBuffer buffer = new RewindableStreamBuffer(new ByteArrayInputStream(data), 8); // Initial fill: "01234567" (bytes 0..7), stream still has "89" remaining buffer.fill(); @@ -207,7 +197,7 @@ public int read(byte[] b, int off, int len) throws IOException { } }; - RewindableStreamBuffer buffer = new RewindableStreamBuffer(shortReadingStream, 8, UPLOAD_URL); + RewindableStreamBuffer buffer = new RewindableStreamBuffer(shortReadingStream, 8); buffer.fill(); // Must greedily fill all 8 bytes despite short reads, and not be marked final yet