Skip to content

feat(gax): include upload-status header in resumable upload command response objects #14420

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

Draft
whowes wants to merge 1 commit into from
Draft

feat(gax): include upload-status header in resumable upload command response objects #14420

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
58 changes: 38 additions & 20 deletions .../gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadChunkCallable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
import com.google.api.core.ApiFuture;
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.rpc.ApiCallContext;
import com.google.api.gax.rpc.ApiExceptionFactory;
import com.google.api.gax.rpc.ClientContext;
Expand All @@ -59,7 +60,6 @@
private static final String UPLOAD_COMMAND_HEADER = "X-Goog-Upload-Command";
private static final String UPLOAD_OFFSET_HEADER = "X-Goog-Upload-Offset";
private static final String UPLOAD_STATUS_HEADER = "X-Goog-Upload-Status";
private static final String STATUS_FINAL = "final";

private static final String COMMAND_UPLOAD = "upload";
private static final String COMMAND_FINALIZE = "finalize";
Expand Down Expand Up @@ -165,7 +165,7 @@

private final ResumableUploadHttpJsonFuture<ChunkUploadResponse<ResponseT>> future;
private final HttpResponseParser<ResponseT> responseParser;
@Nullable private String uploadStatus = null;
private ResumableUploadStatus uploadStatus = ResumableUploadStatus.UNKNOWN;
private String responseBody = "";

private ChunkUploadResponseListener(
Expand All @@ -178,7 +178,9 @@
@Override
public void onHeaders(HttpJsonMetadata responseHeaders) {
Map<String, Object> headers = responseHeaders.getHeaders();
this.uploadStatus = HttpHeadersUtils.getSingleHeader(headers, UPLOAD_STATUS_HEADER);
this.uploadStatus =
ResumableUploadStatus.fromHeader(
HttpHeadersUtils.getSingleHeader(headers, UPLOAD_STATUS_HEADER));
}

@Override
Expand All @@ -192,26 +194,15 @@
public void onClose(int statusCode, HttpJsonMetadata trailers) {
try {
if (statusCode >= 200 && statusCode < 300) {
if (uploadStatus == null) {
future.setException(
ApiExceptionFactory.createException(
"Upload chunk response did not contain valid "
+ UPLOAD_STATUS_HEADER
+ " header",
/* cause= */ null,
HttpJsonStatusCode.of(StatusCode.Code.INTERNAL),
/* retryable= */ false));
return;
}
boolean isComplete = STATUS_FINAL.equalsIgnoreCase(uploadStatus);
ChunkUploadResponse.Builder<ResponseT> chunkResponseBuilder =
ChunkUploadResponse.<ResponseT>newBuilder().setComplete(isComplete);
if (isComplete) {
ResponseT response = null;
if (uploadStatus == ResumableUploadStatus.FINAL) {
InputStream stream =
new ByteArrayInputStream(responseBody.getBytes(StandardCharsets.UTF_8));
chunkResponseBuilder.setResponse(responseParser.parse(stream));
response = responseParser.parse(stream);
}
future.set(chunkResponseBuilder.build());
future.set(ChunkUploadResponse.create(uploadStatus, response));
} else if (uploadStatus == ResumableUploadStatus.FINAL) {
future.setException(createServerRejectionException(statusCode, uploadStatus, trailers));
} else {
Throwable cause = trailers.getException();
future.setException(
Expand All @@ -225,4 +216,31 @@
}
}
}

static Throwable createServerRejectionException(
int statusCode, ResumableUploadStatus uploadStatus, HttpJsonMetadata trailers) {
Throwable cause = trailers.getException();
String message = "Upload " + uploadStatus + " by server with status code: " + statusCode;
if (cause != null && cause.getMessage() != null) {
message += ": " + cause.getMessage();
}
return ApiExceptionFactory.createException(
message, cause, createRejectionStatusCode(statusCode), /* retryable= */ false);
}

private static StatusCode createRejectionStatusCode(int httpStatusCode) {

@blakeli0 blakeli0 Sep 22, 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.

Is the purpose of createRejectionStatusCode to disable possible retries? If yes, I'm not a fan of recreating a new StatusCode with null transportCode just to disable the retries. I think this may hide some problems and make things harder to debug in some cases.

Can we still keep the original statusCode? And pass the header to RetryAlgorithm to decide if we should retry?

@blakeli0 blakeli0 Sep 22, 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.

Separately, I think this "rejection flow" can be done in a separate PR. We can merge this PR to get status header first to unblock other PRs.

StatusCode.Code canonicalCode = HttpJsonStatusCode.of(httpStatusCode).getCode();
return new StatusCode() {
@Override
public Code getCode() {
return canonicalCode;
}

@Override
public @Nullable Integer getTransportCode() {

Check failure on line 240 in sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadChunkCallable.java

View check run for this annotation

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

Fix the incompatibility of the annotation @Nullable to honor @NullMarked at class level of the overridden method. 


See more on https://sonarcloud.io/project/issues?id=googleapis_google-cloud-java_showcase&issues=AaC7jDNdMe4CmcK0m77f&open=AaC7jDNdMe4CmcK0m77f&pullRequest=14420
// null transportCode ensures error is classified as FATAL
return null;
}
};
}
}
26 changes: 16 additions & 10 deletions ...ttpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadQueryStatusCallable.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
import com.google.api.core.ApiFuture;
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.rpc.ApiCallContext;
import com.google.api.gax.rpc.ApiExceptionFactory;
import com.google.api.gax.rpc.ClientContext;
Expand Down Expand Up @@ -63,7 +64,6 @@ class ResumableUploadQueryStatusCallable<ResponseT>
private static final String UPLOAD_COMMAND_HEADER = "X-Goog-Upload-Command";
private static final String UPLOAD_STATUS_HEADER = "X-Goog-Upload-Status";
private static final String UPLOAD_SIZE_RECEIVED_HEADER = "X-Goog-Upload-Size-Received";
private static final String STATUS_FINAL = "final";
private static final String COMMAND_QUERY = "query";

private static final Map<String, List<String>> QUERY_STATUS_HEADERS =
Expand Down Expand Up @@ -182,7 +182,7 @@ private static class QueryStatusResponseListener<ResponseT>

private final ResumableUploadHttpJsonFuture<QueryStatusResponse<ResponseT>> future;
private final HttpResponseParser<ResponseT> responseParser;
@Nullable private String uploadStatus = null;
private ResumableUploadStatus uploadStatus = ResumableUploadStatus.UNKNOWN;
@Nullable private Long committedOffset = null;
@Nullable private Throwable headerParsingException;
private String responseBody = "";
Expand All @@ -197,7 +197,9 @@ private QueryStatusResponseListener(
@Override
public void onHeaders(HttpJsonMetadata responseHeaders) {
Map<String, Object> headers = responseHeaders.getHeaders();
this.uploadStatus = HttpHeadersUtils.getSingleHeader(headers, UPLOAD_STATUS_HEADER);
this.uploadStatus =
ResumableUploadStatus.fromHeader(
HttpHeadersUtils.getSingleHeader(headers, UPLOAD_STATUS_HEADER));
try {
this.committedOffset = parseSizeReceived(responseHeaders);
} catch (Throwable t) {
Expand All @@ -220,19 +222,19 @@ public void onClose(int statusCode, HttpJsonMetadata trailers) {
future.setException(headerParsingException);
return;
}
boolean isComplete = STATUS_FINAL.equalsIgnoreCase(uploadStatus);
if (isComplete) {
QueryStatusResponse.Builder<ResponseT> queryResponseBuilder =
QueryStatusResponse.<ResponseT>newBuilder().setComplete(true);
if (uploadStatus == ResumableUploadStatus.FINAL) {
InputStream stream =
new ByteArrayInputStream(responseBody.getBytes(StandardCharsets.UTF_8));
queryResponseBuilder.setResponse(responseParser.parse(stream));
future.set(queryResponseBuilder.build());
future.set(
QueryStatusResponse.<ResponseT>newBuilder()
.setUploadStatus(uploadStatus)
.setResponse(responseParser.parse(stream))
.build());
} else if (committedOffset != null) {
future.set(
QueryStatusResponse.<ResponseT>newBuilder()
.setComplete(false)
.setCommittedOffset(committedOffset)
.setUploadStatus(uploadStatus)
.build());
} else {
future.setException(
Expand All @@ -244,6 +246,10 @@ public void onClose(int statusCode, HttpJsonMetadata trailers) {
HttpJsonStatusCode.of(StatusCode.Code.INTERNAL),
/* retryable= */ false));
}
} else if (uploadStatus == ResumableUploadStatus.FINAL) {
future.setException(
ResumableUploadChunkCallable.createServerRejectionException(
statusCode, uploadStatus, trailers));
} else {
Throwable cause = trailers.getException();
future.setException(
Expand Down
62 changes: 44 additions & 18 deletions ...httpjson/src/test/java/com/google/api/gax/httpjson/HttpJsonResumableUploadClientTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,13 +45,15 @@
import com.google.api.gax.resumable.QueryStatusRequest;
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.rpc.AbortedException;
import com.google.api.gax.rpc.ApiCallContext;
import com.google.api.gax.rpc.ApiException;
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;
Expand Down Expand Up @@ -260,8 +262,8 @@ void uploadChunk_intermediateChunk_sendsUploadCommandAndReturnsActiveStatus() {

ChunkUploadResponse<String> response = client.uploadChunkCallable().call(request);

assertThat(response.isComplete()).isFalse();
assertThat(response.getResponse()).isNull();
assertThat(response.getUploadStatus()).isEqualTo(ResumableUploadStatus.ACTIVE);

assertThat(transport.capturedUrl).isEqualTo(TEST_UPLOAD_URL);
assertThat(transport.capturedHeaders.get("x-goog-upload-command")).containsExactly("upload");
Expand Down Expand Up @@ -289,9 +291,9 @@ void uploadChunk_finalChunk_sendsUploadFinalizeAndReturnsResponseBody() {

ChunkUploadResponse<String> response = client.uploadChunkCallable().call(request);

assertThat(response.isComplete()).isTrue();
assertThat(response.getResponse())
.isEqualTo("{\"name\":\"uploaded-file.txt\",\"size\":524288}");
assertThat(response.getUploadStatus()).isEqualTo(ResumableUploadStatus.FINAL);

assertThat(transport.capturedHeaders.get("x-goog-upload-command"))
.containsExactly("upload, finalize");
Expand All @@ -318,9 +320,9 @@ void uploadChunk_emptyPayloadFinal_sendsFinalizeCommandAndReturnsResponseBody()

ChunkUploadResponse<String> response = client.uploadChunkCallable().call(request);

assertThat(response.isComplete()).isTrue();
assertThat(response.getResponse())
.isEqualTo("{\"name\":\"uploaded-file.txt\",\"size\":1048576}");
assertThat(response.getUploadStatus()).isEqualTo(ResumableUploadStatus.FINAL);

assertThat(transport.capturedHeaders.get("x-goog-upload-command")).containsExactly("finalize");
assertThat(transport.capturedHeaders).doesNotContainKey("x-goog-upload-offset");
Expand Down Expand Up @@ -379,7 +381,7 @@ void uploadChunk_serverReturnsConflictOrError_throwsException() {
}

@Test
void uploadChunk_missingUploadStatusHeader_throwsInternalException() {
void uploadChunk_missingUploadStatusHeader_returnsUnknownUploadStatusOnHttp200() {
MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse();
httpResponse.setStatusCode(200);

Expand All @@ -391,14 +393,10 @@ void uploadChunk_missingUploadStatusHeader_throwsInternalException() {
.setOffset(0L)
.build();

ExecutionException exception =
assertThrows(
ExecutionException.class, () -> client.uploadChunkCallable().futureCall(request).get());
ChunkUploadResponse<String> response = client.uploadChunkCallable().call(request);

assertThat(exception.getCause()).isInstanceOf(InternalException.class);
assertThat(exception.getCause())
.hasMessageThat()
.contains("Upload chunk response did not contain valid X-Goog-Upload-Status header");
assertThat(response.getResponse()).isNull();
assertThat(response.getUploadStatus()).isEqualTo(ResumableUploadStatus.UNKNOWN);
}

@Test
Expand All @@ -420,10 +418,38 @@ void uploadChunk_serverReturnsFinalStatusOnNon200_marksExceptionNonRetryable() {
assertThrows(
ExecutionException.class, () -> client.uploadChunkCallable().futureCall(request).get());

assertThat(exception.getCause()).isInstanceOf(ApiException.class);
ApiException apiException = (ApiException) exception.getCause();
assertThat(apiException.isRetryable()).isFalse();
assertThat(apiException.getStatusCode().getCode()).isEqualTo(StatusCode.Code.UNAVAILABLE);
assertThat(exception.getCause()).isInstanceOf(UnavailableException.class);
UnavailableException unavailable = (UnavailableException) exception.getCause();
assertThat(unavailable.isRetryable()).isFalse();
assertThat(unavailable.getStatusCode().getCode()).isEqualTo(StatusCode.Code.UNAVAILABLE);
assertThat(unavailable.getStatusCode().getTransportCode()).isNull();
}

@Test
void uploadChunk_serverReturnsFinalStatusOn400_throwsInvalidArgumentException() {
MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse();
httpResponse.setStatusCode(400);
httpResponse.addHeader("X-Goog-Upload-Status", "final");
httpResponse.setContent("{\"error\":{\"message\":\"Invalid chunk format\"}}");

HttpJsonResumableUploadClient<TestRequest, String> client = createClient(httpResponse);
ChunkUploadRequest request =
ChunkUploadRequest.newBuilder()
.setUploadUrl(TEST_UPLOAD_URL)
.setPayload("data".getBytes(StandardCharsets.UTF_8))
.setOffset(0L)
.build();

ExecutionException exception =
assertThrows(
ExecutionException.class, () -> client.uploadChunkCallable().futureCall(request).get());

assertThat(exception.getCause()).isInstanceOf(InvalidArgumentException.class);
InvalidArgumentException invalidArgument = (InvalidArgumentException) exception.getCause();
assertThat(invalidArgument.isRetryable()).isFalse();
assertThat(invalidArgument.getStatusCode().getCode())
.isEqualTo(StatusCode.Code.INVALID_ARGUMENT);
assertThat(invalidArgument.getStatusCode().getTransportCode()).isNull();
}

@Test
Expand All @@ -439,9 +465,9 @@ void queryStatus_activeUpload_returnsCommittedOffset() {

QueryStatusResponse<String> response = client.queryStatusCallable().call(request);

assertThat(response.isComplete()).isFalse();
assertThat(response.getCommittedOffset()).isEqualTo(524288L);
assertThat(response.getResponse()).isNull();
assertThat(response.getUploadStatus()).isEqualTo(ResumableUploadStatus.ACTIVE);

assertThat(transport.capturedHeaders.get("x-goog-upload-command")).containsExactly("query");
}
Expand All @@ -458,10 +484,10 @@ void queryStatus_finalUpload_returnsCompleteAndResponseBody() {

QueryStatusResponse<String> response = client.queryStatusCallable().call(request);

assertThat(response.isComplete()).isTrue();
assertThat(response.getCommittedOffset()).isNull();
assertThat(response.getResponse())
.isEqualTo("{\"name\":\"uploaded-file.txt\",\"size\":1048576}");
assertThat(response.getUploadStatus()).isEqualTo(ResumableUploadStatus.FINAL);
}

@Test
Expand Down
17 changes: 9 additions & 8 deletions ...orm-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ChunkUploadResponse.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,35 +46,36 @@
@AutoValue
public abstract class ChunkUploadResponse<ResponseT> {

/** Whether the overall resumable upload stream has finalized and completed on the server. */
public abstract boolean isComplete();

/**
* The response object returned by the server upon final completion (e.g. metadata of the uploaded
* resource), or {@code null} if the upload is still in progress.
*/
public abstract @Nullable ResponseT getResponse();

/** Returns the status of the upload session returned by the server. */
public abstract ResumableUploadStatus getUploadStatus();

public abstract Builder<ResponseT> toBuilder();

public static <ResponseT> Builder<ResponseT> newBuilder() {
return new AutoValue_ChunkUploadResponse.Builder<ResponseT>().setComplete(false);
return new AutoValue_ChunkUploadResponse.Builder<ResponseT>()
.setUploadStatus(ResumableUploadStatus.ACTIVE);
}

public static <ResponseT> ChunkUploadResponse<ResponseT> create(
boolean isComplete, @Nullable ResponseT response) {
ResumableUploadStatus uploadStatus, @Nullable ResponseT response) {
return new AutoValue_ChunkUploadResponse.Builder<ResponseT>()
.setComplete(isComplete)
.setUploadStatus(uploadStatus)
.setResponse(response)
.build();
}

@AutoValue.Builder
public abstract static class Builder<ResponseT> {
public abstract Builder<ResponseT> setComplete(boolean isComplete);

public abstract Builder<ResponseT> setResponse(@Nullable ResponseT response);

public abstract Builder<ResponseT> setUploadStatus(ResumableUploadStatus uploadStatus);

public abstract ChunkUploadResponse<ResponseT> build();
}
}
Loading
Loading