From 1a0efb3e5a43372e649eb247a9e4af6199e8562a Mon Sep 17 00:00:00 2001 From: whowes Date: Fri, 30 Oct 2026 15:22:22 +0000 Subject: [PATCH] feat(gax): wire-level changes for resuming uploads --- .../ResumableUploadChunkCallable.java | 16 ++-- .../gax/httpjson/ResumableUploadHeaders.java | 85 +++++++++++++++++++ .../ResumableUploadQueryStatusCallable.java | 20 +++-- .../ResumableUploadStartCallable.java | 44 +++------- .../HttpJsonResumableUploadClientTest.java | 33 ++++++- .../gax/resumable/QueryStatusResponse.java | 11 ++- .../gax/resumable/ResumableUploadSession.java | 2 +- 7 files changed, 158 insertions(+), 53 deletions(-) create mode 100644 sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadHeaders.java 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 56c9ffbf98c1..d8eb243cd2ff 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 @@ -29,6 +29,9 @@ */ package com.google.api.gax.httpjson; +import static com.google.api.gax.httpjson.ResumableUploadHeaders.UPLOAD_COMMAND_HEADER; +import static com.google.api.gax.httpjson.ResumableUploadHeaders.UPLOAD_OFFSET_HEADER; + import com.google.api.client.http.HttpMethods; import com.google.api.core.ApiFuture; import com.google.api.gax.resumable.ChunkUploadRequest; @@ -36,6 +39,7 @@ import com.google.api.gax.resumable.ResumableUploadStatus; import com.google.api.gax.rpc.ApiCallContext; import com.google.api.gax.rpc.ClientContext; +import com.google.api.gax.rpc.EndpointContext; import com.google.api.gax.rpc.UnaryCallable; import com.google.api.pathtemplate.PathTemplate; import com.google.common.base.Preconditions; @@ -55,10 +59,6 @@ class ResumableUploadChunkCallable extends UnaryCallable> { - 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 COMMAND_UPLOAD = "upload"; private static final String COMMAND_FINALIZE = "finalize"; private static final String COMMAND_UPLOAD_FINALIZE = "upload, finalize"; @@ -129,6 +129,9 @@ public ApiFuture> futureCall( HttpJsonCallContext.createDefault() .nullToSelf(clientContext.getDefaultCallContext()) .merge(inputContext) + // Session URLs are self-authenticating; omit credentials and reset EndpointContext. + .withCredentials(null) + .withEndpointContext(EndpointContext.getDefaultInstance()) .withExtraHeaders(chunkHeaders); HttpJsonClientCall clientCall = @@ -175,10 +178,7 @@ private ChunkUploadResponseListener( @Override public void onHeaders(HttpJsonMetadata responseHeaders) { - Map headers = responseHeaders.getHeaders(); - this.uploadStatus = - ResumableUploadStatus.fromHeader( - HttpHeadersUtils.getSingleHeader(headers, UPLOAD_STATUS_HEADER)); + this.uploadStatus = ResumableUploadHeaders.parseUploadStatus(responseHeaders); } @Override diff --git a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadHeaders.java b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadHeaders.java new file mode 100644 index 000000000000..2872f8258d3d --- /dev/null +++ b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadHeaders.java @@ -0,0 +1,85 @@ +/* + * 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.httpjson; + +import com.google.api.gax.resumable.ResumableUploadSession; +import com.google.api.gax.resumable.ResumableUploadStatus; +import com.google.api.gax.rpc.ApiExceptionFactory; +import com.google.api.gax.rpc.StatusCode; +import com.google.common.base.Strings; +import org.jspecify.annotations.NullMarked; + +/** Shared header constants and parsing utilities for HTTP/JSON resumable upload wire calls. */ +@NullMarked +final class ResumableUploadHeaders { + + static final String UPLOAD_PROTOCOL_HEADER = "X-Goog-Upload-Protocol"; + static final String UPLOAD_COMMAND_HEADER = "X-Goog-Upload-Command"; + static final String UPLOAD_URL_HEADER = "X-Goog-Upload-URL"; + static final String UPLOAD_GRANULARITY_HEADER = "X-Goog-Upload-Chunk-Granularity"; + static final String UPLOAD_OFFSET_HEADER = "X-Goog-Upload-Offset"; + static final String UPLOAD_STATUS_HEADER = "X-Goog-Upload-Status"; + static final String UPLOAD_SIZE_RECEIVED_HEADER = "X-Goog-Upload-Size-Received"; + + private ResumableUploadHeaders() {} + + static ResumableUploadStatus parseUploadStatus(HttpJsonMetadata responseHeaders) { + return ResumableUploadStatus.fromHeader( + HttpHeadersUtils.getSingleHeader(responseHeaders.getHeaders(), UPLOAD_STATUS_HEADER)); + } + + static int parseChunkGranularity(HttpJsonMetadata responseHeaders) { + String granularityStr = + HttpHeadersUtils.getSingleHeader(responseHeaders.getHeaders(), UPLOAD_GRANULARITY_HEADER); + if (Strings.isNullOrEmpty(granularityStr)) { + return ResumableUploadSession.DEFAULT_CHUNK_GRANULARITY; + } + try { + int parsed = Integer.parseInt(granularityStr); + if (parsed <= 0) { + throw ApiExceptionFactory.createException( + "Response contained non-positive " + + UPLOAD_GRANULARITY_HEADER + + " header: " + + granularityStr, + /* cause= */ null, + HttpJsonStatusCode.of(StatusCode.Code.INTERNAL), + /* retryable= */ false); + } + return parsed; + } catch (NumberFormatException e) { + throw ApiExceptionFactory.createException( + "Response contained invalid " + UPLOAD_GRANULARITY_HEADER + " header: " + granularityStr, + e, + HttpJsonStatusCode.of(StatusCode.Code.INTERNAL), + /* retryable= */ 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..92f8f3d6789e 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 @@ -29,14 +29,19 @@ */ package com.google.api.gax.httpjson; +import static com.google.api.gax.httpjson.ResumableUploadHeaders.UPLOAD_COMMAND_HEADER; +import static com.google.api.gax.httpjson.ResumableUploadHeaders.UPLOAD_SIZE_RECEIVED_HEADER; + import com.google.api.client.http.HttpMethods; 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.ResumableUploadSession; 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; +import com.google.api.gax.rpc.EndpointContext; import com.google.api.gax.rpc.StatusCode; import com.google.api.gax.rpc.UnaryCallable; import com.google.api.pathtemplate.PathTemplate; @@ -61,9 +66,6 @@ class ResumableUploadQueryStatusCallable extends UnaryCallable> { - 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 COMMAND_QUERY = "query"; private static final Map> QUERY_STATUS_HEADERS = @@ -119,6 +121,9 @@ public ApiFuture> futureCall( HttpJsonCallContext.createDefault() .nullToSelf(clientContext.getDefaultCallContext()) .merge(inputContext) + // Session URLs are self-authenticating; omit credentials and reset EndpointContext. + .withCredentials(null) + .withEndpointContext(EndpointContext.getDefaultInstance()) .withExtraHeaders(QUERY_STATUS_HEADERS); HttpJsonClientCall clientCall = @@ -183,6 +188,7 @@ private static class QueryStatusResponseListener private final ResumableUploadHttpJsonFuture> future; private final HttpResponseParser responseParser; private ResumableUploadStatus uploadStatus = ResumableUploadStatus.UNKNOWN; + private int chunkGranularity = ResumableUploadSession.DEFAULT_CHUNK_GRANULARITY; @Nullable private Long committedOffset = null; @Nullable private Throwable headerParsingException; private String responseBody = ""; @@ -196,12 +202,10 @@ private QueryStatusResponseListener( @Override public void onHeaders(HttpJsonMetadata responseHeaders) { - Map headers = responseHeaders.getHeaders(); - this.uploadStatus = - ResumableUploadStatus.fromHeader( - HttpHeadersUtils.getSingleHeader(headers, UPLOAD_STATUS_HEADER)); + this.uploadStatus = ResumableUploadHeaders.parseUploadStatus(responseHeaders); try { this.committedOffset = parseSizeReceived(responseHeaders); + this.chunkGranularity = ResumableUploadHeaders.parseChunkGranularity(responseHeaders); } catch (Throwable t) { this.headerParsingException = t; } @@ -229,12 +233,14 @@ public void onClose(int statusCode, HttpJsonMetadata trailers) { QueryStatusResponse.newBuilder() .setUploadStatus(uploadStatus) .setResponse(responseParser.parse(stream)) + .setChunkGranularity(chunkGranularity) .build()); } else if (committedOffset != null) { future.set( QueryStatusResponse.newBuilder() .setCommittedOffset(committedOffset) .setUploadStatus(uploadStatus) + .setChunkGranularity(chunkGranularity) .build()); } else { future.setException( diff --git a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadStartCallable.java b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadStartCallable.java index 179bd5dcca32..07d959e3650d 100644 --- a/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadStartCallable.java +++ b/sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadStartCallable.java @@ -29,6 +29,10 @@ */ package com.google.api.gax.httpjson; +import static com.google.api.gax.httpjson.ResumableUploadHeaders.UPLOAD_COMMAND_HEADER; +import static com.google.api.gax.httpjson.ResumableUploadHeaders.UPLOAD_PROTOCOL_HEADER; +import static com.google.api.gax.httpjson.ResumableUploadHeaders.UPLOAD_URL_HEADER; + import com.google.api.core.ApiFuture; import com.google.api.gax.resumable.ResumableUploadSession; import com.google.api.gax.rpc.ApiCallContext; @@ -52,11 +56,6 @@ class ResumableUploadStartCallable extends UnaryCallable { - private static final String UPLOAD_PROTOCOL_HEADER = "X-Goog-Upload-Protocol"; - private static final String UPLOAD_COMMAND_HEADER = "X-Goog-Upload-Command"; - private static final String UPLOAD_URL_HEADER = "X-Goog-Upload-URL"; - private static final String UPLOAD_GRANULARITY_HEADER = "X-Goog-Upload-Chunk-Granularity"; - private static final Map> START_UPLOAD_HEADERS = ImmutableMap.of( UPLOAD_PROTOCOL_HEADER, ImmutableList.of("resumable"), @@ -108,7 +107,7 @@ static UnaryCallable create( /** A listener that parses HTTP response headers to produce a {@link ResumableUploadSession}. */ private static class StartUploadResponseListener extends HttpJsonClientCall.Listener { private final ResumableUploadHttpJsonFuture future; - private int chunkGranularity = 1; + private int chunkGranularity = ResumableUploadSession.DEFAULT_CHUNK_GRANULARITY; @Nullable private String uploadUrl; @Nullable private Throwable headerParsingException; @@ -119,39 +118,16 @@ private StartUploadResponseListener( @Override public void onHeaders(HttpJsonMetadata responseHeaders) { - Map headers = responseHeaders.getHeaders(); - - String url = HttpHeadersUtils.getSingleHeader(headers, UPLOAD_URL_HEADER); + String url = + HttpHeadersUtils.getSingleHeader(responseHeaders.getHeaders(), UPLOAD_URL_HEADER); if (!Strings.isNullOrEmpty(url)) { this.uploadUrl = url; } - String granularityStr = HttpHeadersUtils.getSingleHeader(headers, UPLOAD_GRANULARITY_HEADER); - if (Strings.isNullOrEmpty(granularityStr)) { - return; - } - try { - int parsed = Integer.parseInt(granularityStr); - if (parsed <= 0) { - this.headerParsingException = - ApiExceptionFactory.createException( - "Start upload response contained non-positive chunk granularity header: " - + granularityStr, - /* cause= */ null, - HttpJsonStatusCode.of(StatusCode.Code.INTERNAL), - /* retryable= */ false); - } else { - this.chunkGranularity = parsed; - } - } catch (NumberFormatException e) { - this.headerParsingException = - ApiExceptionFactory.createException( - "Start upload response contained invalid chunk granularity header: " - + granularityStr, - e, - HttpJsonStatusCode.of(StatusCode.Code.INTERNAL), - /* retryable= */ false); + this.chunkGranularity = ResumableUploadHeaders.parseChunkGranularity(responseHeaders); + } catch (Throwable t) { + this.headerParsingException = t; } } 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 076a42e30bd1..ed9039877ef6 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 @@ -57,6 +57,9 @@ import com.google.api.gax.rpc.StatusCode; import com.google.api.gax.rpc.UnaryCallSettings; import com.google.api.pathtemplate.PathTemplate; +import com.google.auth.Credentials; +import com.google.auth.oauth2.AccessToken; +import com.google.auth.oauth2.OAuth2Credentials; import com.google.common.base.Strings; import java.io.IOException; import java.nio.charset.StandardCharsets; @@ -161,7 +164,7 @@ void startUpload_nonPositiveChunkGranularityHeader_throwsException() { assertThat(exception.getCause()).isInstanceOf(InternalException.class); assertThat(exception.getCause()) .hasMessageThat() - .contains("Start upload response contained non-positive chunk granularity header: -256"); + .contains("Response contained non-positive X-Goog-Upload-Chunk-Granularity header: -256"); } @Test @@ -205,6 +208,7 @@ void startUpload_withPayloadAndQueryParams_sendsCorrectRequest() { assertThat(transport.capturedHeaders.get("x-goog-upload-protocol")) .containsExactly("resumable"); assertThat(transport.capturedHeaders.get("x-goog-upload-command")).containsExactly("start"); + assertThat(transport.capturedHeaders.get("authorization")).containsExactly("Bearer test-token"); } @Test @@ -336,6 +340,7 @@ void uploadChunk_intermediateChunk_sendsUploadCommandAndReturnsActiveStatus() { assertThat(transport.capturedUrl).isEqualTo(TEST_UPLOAD_URL); assertThat(transport.capturedHeaders.get("x-goog-upload-command")).containsExactly("upload"); assertThat(transport.capturedHeaders.get("x-goog-upload-offset")).containsExactly("0"); + assertThat(transport.capturedHeaders).doesNotContainKey("authorization"); } @Test @@ -512,6 +517,25 @@ void queryStatus_activeUpload_returnsCommittedOffset() { assertThat(response.getUploadStatus()).isEqualTo(ResumableUploadStatus.ACTIVE); assertThat(transport.capturedHeaders.get("x-goog-upload-command")).containsExactly("query"); + assertThat(transport.capturedHeaders).doesNotContainKey("authorization"); + } + + @Test + void queryStatus_withChunkGranularity_returnsGranularity() { + MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse(); + httpResponse.setStatusCode(200); + httpResponse.addHeader("X-Goog-Upload-Status", "active"); + httpResponse.addHeader("X-Goog-Upload-Size-Received", "524288"); + httpResponse.addHeader("X-Goog-Upload-Chunk-Granularity", "262144"); + + HttpJsonResumableUploadClient client = createClient(httpResponse); + QueryStatusRequest request = QueryStatusRequest.create(TEST_UPLOAD_URL); + + QueryStatusResponse response = client.queryStatusCallable().call(request); + + assertThat(response.getCommittedOffset()).isEqualTo(524288L); + assertThat(response.getChunkGranularity()).isEqualTo(262144); + assertThat(response.getUploadStatus()).isEqualTo(ResumableUploadStatus.ACTIVE); } @Test @@ -622,10 +646,15 @@ private static HttpJsonResumableUploadClient createClient( .setHttpTransport(transport) .build(); + Credentials credentials = OAuth2Credentials.create(new AccessToken("test-token", null)); + HttpJsonCallContext defaultCallContext = + HttpJsonCallContext.createDefault().withChannel(channel).withCredentials(credentials); + ClientContext clientContext = ClientContext.newBuilder() .setTransportChannel(HttpJsonTransportChannel.create(channel)) - .setDefaultCallContext(HttpJsonCallContext.createDefault().withChannel(channel)) + .setDefaultCallContext(defaultCallContext) + .setCredentials(credentials) .build(); return HttpJsonResumableUploadClient.create( diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/QueryStatusResponse.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/QueryStatusResponse.java index 39c825269157..78908cf666e5 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/QueryStatusResponse.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/QueryStatusResponse.java @@ -65,11 +65,18 @@ public abstract class QueryStatusResponse { /** Returns the status of the upload session returned by the server. */ public abstract ResumableUploadStatus getUploadStatus(); + /** + * Returns the byte alignment granularity required by the server for chunk uploads, or {@code 1} + * if no alignment restriction was returned. + */ + public abstract int getChunkGranularity(); + public abstract Builder toBuilder(); public static Builder newBuilder() { return new AutoValue_QueryStatusResponse.Builder() - .setUploadStatus(ResumableUploadStatus.ACTIVE); + .setUploadStatus(ResumableUploadStatus.ACTIVE) + .setChunkGranularity(ResumableUploadSession.DEFAULT_CHUNK_GRANULARITY); } @AutoValue.Builder @@ -80,6 +87,8 @@ public abstract static class Builder { public abstract Builder setUploadStatus(ResumableUploadStatus uploadStatus); + public abstract Builder setChunkGranularity(int chunkGranularity); + public abstract QueryStatusResponse build(); } } diff --git a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadSession.java b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadSession.java index 1e6e9d359538..dd5d817a3864 100644 --- a/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadSession.java +++ b/sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ResumableUploadSession.java @@ -41,7 +41,7 @@ @AutoValue public abstract class ResumableUploadSession { - private static final int DEFAULT_CHUNK_GRANULARITY = 1; + public static final int DEFAULT_CHUNK_GRANULARITY = 1; /** Returns the server-provided URL to which data uploads are directed. */ public abstract String getUploadUrl();