Skip to content

feat(gax): standardize mapping terminal resumable upload exceptions to ApiException #14624

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
whowes wants to merge 1 commit into
base: main
Choose a base branch
from
+148 −65
Open

feat(gax): standardize mapping terminal resumable upload exceptions to ApiException #14624

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 5 additions & 9 deletions ...va/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadChunkCoordinator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,7 +99,7 @@ final class ResumableUploadChunkCoordinator<ResponseT> {
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<ResponseT> getFuture() {
Expand Down Expand Up @@ -150,8 +150,7 @@ public void onSuccess(ChunkUploadResponse<ResponseT> 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);
Expand Down Expand Up @@ -263,8 +262,7 @@ private boolean tryRegisterInFlightFuture(Future<?> future) {
private void handleQueryResponse(QueryStatusResponse<ResponseT> 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);
Expand All @@ -274,8 +272,7 @@ private void handleQueryResponse(QueryStatusResponse<ResponseT> 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);
Expand All @@ -290,8 +287,7 @@ private void handleChunkResponse(ChunkUploadResponse<ResponseT> 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();
Expand Down
71 changes: 49 additions & 22 deletions ...orm-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadFutureImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -212,14 +212,15 @@
}

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) {
Expand All @@ -229,7 +230,7 @@
}
inFlightFuture = null;
}
closePayload();
closePayload(null);
resultFuture.set(result);
}

Expand All @@ -243,19 +244,44 @@
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());
Comment thread
whowes marked this conversation as resolved.
if (uploadSessionUrl != null) {
message = message + " (upload URL: " + uploadSessionUrl + ")";

@blakeli0 blakeli0 Oct 12, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We might missed this in previous PRs. I think the upload URL contains an upload id, and anyone with the id can upload without authentication? If that's the case, I think it would be better to not include the full url in the log.

@whowes whowes Oct 12, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This did come up in an earlier PR - revisiting this I think that we're required to surface the full URL. From internal cross-language design docs:

"The Client Libraries should include the upload URL as part of the errors it returns to the end-user when resumable upload-related errors occur."

and

"The upload URL can be configured (in the server configuration) to be completely arbitrary, so we cannot guarantee that we can parse the upload ID. The URL will have to be surfaced in its entirety."

}
// 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) {

Check warning on line 281 in sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/ResumableUploadFutureImpl.java

View check run for this annotation

SonarQubeCloud / [gapic-generator-java-root] SonarCloud Code Analysis

Catch Exception instead of Throwable. 


See more on https://sonarcloud.io/project/issues?id=googleapis_google-cloud-java_showcase&issues=AaEjHPEMaIdfEDPpYg6d&open=AaEjHPEMaIdfEDPpYg6d&pullRequest=14624
if (primaryException != null) {
primaryException.addSuppressed(e);
}
}
}
Comment thread
whowes marked this conversation as resolved.

Expand Down Expand Up @@ -283,7 +309,7 @@
}
inFlight.cancel(mayInterruptIfRunning);
progressTracker.onFailed();
closePayload();
closePayload(null);
return cancelled;
}

Expand All @@ -308,16 +334,17 @@
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;
}
};
}
}
15 changes: 6 additions & 9 deletions ...atform-java/gax-java/gax/src/main/java/com/google/api/gax/rpc/RewindableStreamBuffer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,19 +46,17 @@ final class RewindableStreamBuffer {

private final InputStream inputStream;
private final int chunkSize;
private final String uploadUrl;
private final byte[] buffer;

private long bufferBaseOffset;
private int payloadLength;
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;
Expand Down Expand Up @@ -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);
Expand Down
83 changes: 78 additions & 5 deletions ...va/gax-java/gax/src/test/java/com/google/api/gax/rpc/ResumableUploadCallableImplTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
}
Expand All @@ -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");
}

Expand Down Expand Up @@ -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");
Expand Down Expand Up @@ -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");
}

Expand Down Expand Up @@ -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");
Expand Down Expand Up @@ -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);
}
Expand Down Expand Up @@ -1259,6 +1261,77 @@ List<ResumableUploadProgress> 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<String> 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<String> 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<String> 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;
Expand Down
Loading
Loading