Skip to content

Commit c9676ea

Browse files
whoweswhowes
committed
feat(gax): wire-level changes for resuming uploads
1 parent 1bc507f commit c9676ea

7 files changed

Lines changed: 144 additions & 78 deletions

File tree

‎sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadChunkCallable.java‎

Lines changed: 4 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,9 @@
2929
*/
3030
package com.google.api.gax.httpjson;
3131

32+
import static com.google.api.gax.httpjson.ResumableUploadHeaders.UPLOAD_COMMAND_HEADER;
33+
import static com.google.api.gax.httpjson.ResumableUploadHeaders.UPLOAD_OFFSET_HEADER;
34+
3235
import com.google.api.client.http.HttpMethods;
3336
import com.google.api.core.ApiFuture;
3437
import com.google.api.gax.resumable.ChunkUploadRequest;
@@ -55,10 +58,6 @@
5558
class ResumableUploadChunkCallable<ResponseT>
5659
extends UnaryCallable<ChunkUploadRequest, ChunkUploadResponse<ResponseT>> {
5760

58-
private static final String UPLOAD_COMMAND_HEADER = "X-Goog-Upload-Command";
59-
private static final String UPLOAD_OFFSET_HEADER = "X-Goog-Upload-Offset";
60-
private static final String UPLOAD_STATUS_HEADER = "X-Goog-Upload-Status";
61-
6261
private static final String COMMAND_UPLOAD = "upload";
6362
private static final String COMMAND_FINALIZE = "finalize";
6463
private static final String COMMAND_UPLOAD_FINALIZE = "upload, finalize";
@@ -175,10 +174,7 @@ private ChunkUploadResponseListener(
175174

176175
@Override
177176
public void onHeaders(HttpJsonMetadata responseHeaders) {
178-
Map<String, Object> headers = responseHeaders.getHeaders();
179-
this.uploadStatus =
180-
ResumableUploadStatus.fromHeader(
181-
HttpHeadersUtils.getSingleHeader(headers, UPLOAD_STATUS_HEADER));
177+
this.uploadStatus = ResumableUploadHeaders.parseUploadStatus(responseHeaders);
182178
}
183179

184180
@Override

‎sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadHeaders.java‎

Lines changed: 78 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,78 @@
1+
/*
2+
* Copyright 2026 Google LLC
3+
*
4+
* Redistribution and use in source and binary forms, with or without
5+
* modification, are permitted provided that the following conditions are
6+
* met:
7+
*
8+
* * Redistributions of source code must retain the above copyright
9+
* notice, this list of conditions and the following disclaimer.
10+
* * Redistributions in binary form must reproduce the above
11+
* copyright notice, this list of conditions and the following disclaimer
12+
* in the documentation and/or other materials provided with the
13+
* distribution.
14+
* * Neither the name of Google LLC nor the names of its
15+
* contributors may be used to endorse or promote products derived from
16+
* this software without specific prior written permission.
17+
*
18+
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
19+
* "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
20+
* LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
21+
* A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
22+
* OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
23+
* SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
24+
* LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
25+
* DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
26+
* THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
27+
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
28+
* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
29+
*/
30+
package com.google.api.gax.httpjson;
31+
32+
import com.google.api.gax.resumable.ResumableUploadSession;
33+
import com.google.api.gax.resumable.ResumableUploadStatus;
34+
import com.google.common.base.Strings;
35+
import org.jspecify.annotations.NullMarked;
36+
37+
/** Shared header constants and parsing utilities for HTTP/JSON resumable upload wire calls. */
38+
@NullMarked
39+
final class ResumableUploadHeaders {
40+
41+
static final String UPLOAD_PROTOCOL_HEADER = "X-Goog-Upload-Protocol";
42+
static final String UPLOAD_COMMAND_HEADER = "X-Goog-Upload-Command";
43+
static final String UPLOAD_URL_HEADER = "X-Goog-Upload-URL";
44+
static final String UPLOAD_GRANULARITY_HEADER = "X-Goog-Upload-Chunk-Granularity";
45+
static final String UPLOAD_OFFSET_HEADER = "X-Goog-Upload-Offset";
46+
static final String UPLOAD_STATUS_HEADER = "X-Goog-Upload-Status";
47+
static final String UPLOAD_SIZE_RECEIVED_HEADER = "X-Goog-Upload-Size-Received";
48+
49+
private ResumableUploadHeaders() {}
50+
51+
static ResumableUploadStatus parseUploadStatus(HttpJsonMetadata responseHeaders) {
52+
return ResumableUploadStatus.fromHeader(
53+
HttpHeadersUtils.getSingleHeader(responseHeaders.getHeaders(), UPLOAD_STATUS_HEADER));
54+
}
55+
56+
static int parseChunkGranularity(HttpJsonMetadata responseHeaders) {
57+
String granularityStr =
58+
HttpHeadersUtils.getSingleHeader(responseHeaders.getHeaders(), UPLOAD_GRANULARITY_HEADER);
59+
if (Strings.isNullOrEmpty(granularityStr)) {
60+
return ResumableUploadSession.DEFAULT_CHUNK_GRANULARITY;
61+
}
62+
try {
63+
int parsed = Integer.parseInt(granularityStr);
64+
if (parsed <= 0) {
65+
throw new IllegalStateException(
66+
"Response contained non-positive "
67+
+ UPLOAD_GRANULARITY_HEADER
68+
+ " header: "
69+
+ granularityStr);
70+
}
71+
return parsed;
72+
} catch (NumberFormatException e) {
73+
throw new IllegalStateException(
74+
"Response contained invalid " + UPLOAD_GRANULARITY_HEADER + " header: " + granularityStr,
75+
e);
76+
}
77+
}
78+
}

‎sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadQueryStatusCallable.java‎

Lines changed: 16 additions & 23 deletions
Original file line numberDiff line numberDiff line change
@@ -29,15 +29,17 @@
2929
*/
3030
package com.google.api.gax.httpjson;
3131

32+
import static com.google.api.gax.httpjson.ResumableUploadHeaders.UPLOAD_COMMAND_HEADER;
33+
import static com.google.api.gax.httpjson.ResumableUploadHeaders.UPLOAD_SIZE_RECEIVED_HEADER;
34+
3235
import com.google.api.client.http.HttpMethods;
3336
import com.google.api.core.ApiFuture;
3437
import com.google.api.gax.resumable.QueryStatusRequest;
3538
import com.google.api.gax.resumable.QueryStatusResponse;
39+
import com.google.api.gax.resumable.ResumableUploadSession;
3640
import com.google.api.gax.resumable.ResumableUploadStatus;
3741
import com.google.api.gax.rpc.ApiCallContext;
38-
import com.google.api.gax.rpc.ApiExceptionFactory;
3942
import com.google.api.gax.rpc.ClientContext;
40-
import com.google.api.gax.rpc.StatusCode;
4143
import com.google.api.gax.rpc.UnaryCallable;
4244
import com.google.api.pathtemplate.PathTemplate;
4345
import com.google.common.base.Preconditions;
@@ -61,9 +63,6 @@
6163
class ResumableUploadQueryStatusCallable<ResponseT>
6264
extends UnaryCallable<QueryStatusRequest, QueryStatusResponse<ResponseT>> {
6365

64-
private static final String UPLOAD_COMMAND_HEADER = "X-Goog-Upload-Command";
65-
private static final String UPLOAD_STATUS_HEADER = "X-Goog-Upload-Status";
66-
private static final String UPLOAD_SIZE_RECEIVED_HEADER = "X-Goog-Upload-Size-Received";
6766
private static final String COMMAND_QUERY = "query";
6867

6968
private static final Map<String, List<String>> QUERY_STATUS_HEADERS =
@@ -152,25 +151,20 @@ private static Long parseSizeReceived(HttpJsonMetadata responseHeaders) {
152151
try {
153152
long parsed = Long.parseLong(sizeReceivedStr);
154153
if (parsed < 0) {
155-
throw ApiExceptionFactory.createException(
154+
throw new IllegalStateException(
156155
"Response contained negative "
157156
+ UPLOAD_SIZE_RECEIVED_HEADER
158157
+ " header: "
159-
+ sizeReceivedStr,
160-
/* cause= */ null,
161-
HttpJsonStatusCode.of(StatusCode.Code.INTERNAL),
162-
/* retryable= */ false);
158+
+ sizeReceivedStr);
163159
}
164160
return parsed;
165161
} catch (NumberFormatException e) {
166-
throw ApiExceptionFactory.createException(
162+
throw new IllegalStateException(
167163
"Response contained invalid "
168164
+ UPLOAD_SIZE_RECEIVED_HEADER
169165
+ " header: "
170166
+ sizeReceivedStr,
171-
e,
172-
HttpJsonStatusCode.of(StatusCode.Code.INTERNAL),
173-
/* retryable= */ false);
167+
e);
174168
}
175169
}
176170
return null;
@@ -183,6 +177,7 @@ private static class QueryStatusResponseListener<ResponseT>
183177
private final ResumableUploadHttpJsonFuture<QueryStatusResponse<ResponseT>> future;
184178
private final HttpResponseParser<ResponseT> responseParser;
185179
private ResumableUploadStatus uploadStatus = ResumableUploadStatus.UNKNOWN;
180+
private int chunkGranularity = ResumableUploadSession.DEFAULT_CHUNK_GRANULARITY;
186181
@Nullable private Long committedOffset = null;
187182
@Nullable private Throwable headerParsingException;
188183
private String responseBody = "";
@@ -196,12 +191,10 @@ private QueryStatusResponseListener(
196191

197192
@Override
198193
public void onHeaders(HttpJsonMetadata responseHeaders) {
199-
Map<String, Object> headers = responseHeaders.getHeaders();
200-
this.uploadStatus =
201-
ResumableUploadStatus.fromHeader(
202-
HttpHeadersUtils.getSingleHeader(headers, UPLOAD_STATUS_HEADER));
194+
this.uploadStatus = ResumableUploadHeaders.parseUploadStatus(responseHeaders);
203195
try {
204196
this.committedOffset = parseSizeReceived(responseHeaders);
197+
this.chunkGranularity = ResumableUploadHeaders.parseChunkGranularity(responseHeaders);
205198
} catch (Throwable t) {
206199
this.headerParsingException = t;
207200
}
@@ -228,23 +221,23 @@ public void onClose(int statusCode, HttpJsonMetadata trailers) {
228221
future.set(
229222
QueryStatusResponse.<ResponseT>newBuilder()
230223
.setUploadStatus(uploadStatus)
224+
.setCommittedOffset(committedOffset)
231225
.setResponse(responseParser.parse(stream))
226+
.setChunkGranularity(chunkGranularity)
232227
.build());
233228
} else if (committedOffset != null) {
234229
future.set(
235230
QueryStatusResponse.<ResponseT>newBuilder()
236231
.setCommittedOffset(committedOffset)
237232
.setUploadStatus(uploadStatus)
233+
.setChunkGranularity(chunkGranularity)
238234
.build());
239235
} else {
240236
future.setException(
241-
ApiExceptionFactory.createException(
237+
new IllegalStateException(
242238
"Query status response did not contain valid "
243239
+ UPLOAD_SIZE_RECEIVED_HEADER
244-
+ " header",
245-
/* cause= */ null,
246-
HttpJsonStatusCode.of(StatusCode.Code.INTERNAL),
247-
/* retryable= */ false));
240+
+ " header"));
248241
}
249242
} else {
250243
Throwable cause = trailers.getException();

‎sdk-platform-java/gax-java/gax-httpjson/src/main/java/com/google/api/gax/httpjson/ResumableUploadStartCallable.java‎

Lines changed: 10 additions & 34 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,10 @@
2929
*/
3030
package com.google.api.gax.httpjson;
3131

32+
import static com.google.api.gax.httpjson.ResumableUploadHeaders.UPLOAD_COMMAND_HEADER;
33+
import static com.google.api.gax.httpjson.ResumableUploadHeaders.UPLOAD_PROTOCOL_HEADER;
34+
import static com.google.api.gax.httpjson.ResumableUploadHeaders.UPLOAD_URL_HEADER;
35+
3236
import com.google.api.core.ApiFuture;
3337
import com.google.api.gax.resumable.ResumableUploadSession;
3438
import com.google.api.gax.rpc.ApiCallContext;
@@ -52,11 +56,6 @@
5256
class ResumableUploadStartCallable<RequestT>
5357
extends UnaryCallable<RequestT, ResumableUploadSession> {
5458

55-
private static final String UPLOAD_PROTOCOL_HEADER = "X-Goog-Upload-Protocol";
56-
private static final String UPLOAD_COMMAND_HEADER = "X-Goog-Upload-Command";
57-
private static final String UPLOAD_URL_HEADER = "X-Goog-Upload-URL";
58-
private static final String UPLOAD_GRANULARITY_HEADER = "X-Goog-Upload-Chunk-Granularity";
59-
6059
private static final Map<String, List<String>> START_UPLOAD_HEADERS =
6160
ImmutableMap.of(
6261
UPLOAD_PROTOCOL_HEADER, ImmutableList.of("resumable"),
@@ -108,7 +107,7 @@ static <RequestT> UnaryCallable<RequestT, ResumableUploadSession> create(
108107
/** A listener that parses HTTP response headers to produce a {@link ResumableUploadSession}. */
109108
private static class StartUploadResponseListener extends HttpJsonClientCall.Listener<String> {
110109
private final ResumableUploadHttpJsonFuture<ResumableUploadSession> future;
111-
private int chunkGranularity = 1;
110+
private int chunkGranularity = ResumableUploadSession.DEFAULT_CHUNK_GRANULARITY;
112111
@Nullable private String uploadUrl;
113112
@Nullable private Throwable headerParsingException;
114113

@@ -119,39 +118,16 @@ private StartUploadResponseListener(
119118

120119
@Override
121120
public void onHeaders(HttpJsonMetadata responseHeaders) {
122-
Map<String, Object> headers = responseHeaders.getHeaders();
123-
124-
String url = HttpHeadersUtils.getSingleHeader(headers, UPLOAD_URL_HEADER);
121+
String url =
122+
HttpHeadersUtils.getSingleHeader(responseHeaders.getHeaders(), UPLOAD_URL_HEADER);
125123
if (!Strings.isNullOrEmpty(url)) {
126124
this.uploadUrl = url;
127125
}
128126

129-
String granularityStr = HttpHeadersUtils.getSingleHeader(headers, UPLOAD_GRANULARITY_HEADER);
130-
if (Strings.isNullOrEmpty(granularityStr)) {
131-
return;
132-
}
133-
134127
try {
135-
int parsed = Integer.parseInt(granularityStr);
136-
if (parsed <= 0) {
137-
this.headerParsingException =
138-
ApiExceptionFactory.createException(
139-
"Start upload response contained non-positive chunk granularity header: "
140-
+ granularityStr,
141-
/* cause= */ null,
142-
HttpJsonStatusCode.of(StatusCode.Code.INTERNAL),
143-
/* retryable= */ false);
144-
} else {
145-
this.chunkGranularity = parsed;
146-
}
147-
} catch (NumberFormatException e) {
148-
this.headerParsingException =
149-
ApiExceptionFactory.createException(
150-
"Start upload response contained invalid chunk granularity header: "
151-
+ granularityStr,
152-
e,
153-
HttpJsonStatusCode.of(StatusCode.Code.INTERNAL),
154-
/* retryable= */ false);
128+
this.chunkGranularity = ResumableUploadHeaders.parseChunkGranularity(responseHeaders);
129+
} catch (Throwable t) {
130+
this.headerParsingException = t;
155131
}
156132
}
157133

‎sdk-platform-java/gax-java/gax-httpjson/src/test/java/com/google/api/gax/httpjson/HttpJsonResumableUploadClientTest.java‎

Lines changed: 25 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,7 @@
5858
import com.google.api.gax.rpc.UnaryCallSettings;
5959
import com.google.api.pathtemplate.PathTemplate;
6060
import com.google.common.base.Strings;
61+
import com.google.common.base.Throwables;
6162
import java.io.IOException;
6263
import java.nio.charset.StandardCharsets;
6364
import java.time.Duration;
@@ -140,8 +141,7 @@ void startUpload_malformedChunkGranularityHeader_throwsException() {
140141
assertThrows(
141142
ExecutionException.class, () -> client.startUploadCallable().futureCall(request).get());
142143

143-
assertThat(exception.getCause()).isInstanceOf(InternalException.class);
144-
assertThat(exception.getCause().getCause()).isInstanceOf(NumberFormatException.class);
144+
assertThat(Throwables.getRootCause(exception)).isInstanceOf(NumberFormatException.class);
145145
}
146146

147147
@Test
@@ -158,10 +158,9 @@ void startUpload_nonPositiveChunkGranularityHeader_throwsException() {
158158
assertThrows(
159159
ExecutionException.class, () -> client.startUploadCallable().futureCall(request).get());
160160

161-
assertThat(exception.getCause()).isInstanceOf(InternalException.class);
162-
assertThat(exception.getCause())
161+
assertThat(Throwables.getRootCause(exception))
163162
.hasMessageThat()
164-
.contains("Start upload response contained non-positive chunk granularity header: -256");
163+
.contains("Response contained non-positive X-Goog-Upload-Chunk-Granularity header: -256");
165164
}
166165

167166
@Test
@@ -514,6 +513,24 @@ void queryStatus_activeUpload_returnsCommittedOffset() {
514513
assertThat(transport.capturedHeaders.get("x-goog-upload-command")).containsExactly("query");
515514
}
516515

516+
@Test
517+
void queryStatus_withChunkGranularity_returnsGranularity() {
518+
MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse();
519+
httpResponse.setStatusCode(200);
520+
httpResponse.addHeader("X-Goog-Upload-Status", "active");
521+
httpResponse.addHeader("X-Goog-Upload-Size-Received", "524288");
522+
httpResponse.addHeader("X-Goog-Upload-Chunk-Granularity", "262144");
523+
524+
HttpJsonResumableUploadClient<TestRequest, String> client = createClient(httpResponse);
525+
QueryStatusRequest request = QueryStatusRequest.create(TEST_UPLOAD_URL);
526+
527+
QueryStatusResponse<String> response = client.queryStatusCallable().call(request);
528+
529+
assertThat(response.getCommittedOffset()).isEqualTo(524288L);
530+
assertThat(response.getChunkGranularity()).isEqualTo(262144);
531+
assertThat(response.getUploadStatus()).isEqualTo(ResumableUploadStatus.ACTIVE);
532+
}
533+
517534
@Test
518535
void queryStatus_finalUpload_returnsCompleteAndResponseBody() {
519536
MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse();
@@ -590,8 +607,7 @@ void queryStatus_missingOrMalformedSizeReceivedHeader_throwsException() {
590607
assertThrows(
591608
ExecutionException.class,
592609
() -> missingClient.queryStatusCallable().futureCall(request).get());
593-
assertThat(missingException.getCause()).isInstanceOf(InternalException.class);
594-
assertThat(missingException.getCause())
610+
assertThat(Throwables.getRootCause(missingException))
595611
.hasMessageThat()
596612
.contains("Query status response did not contain valid X-Goog-Upload-Size-Received header");
597613

@@ -607,10 +623,8 @@ void queryStatus_missingOrMalformedSizeReceivedHeader_throwsException() {
607623
assertThrows(
608624
ExecutionException.class,
609625
() -> malformedClient.queryStatusCallable().futureCall(request).get());
610-
assertThat(malformedException.getCause()).isInstanceOf(InternalException.class);
611-
assertThat(malformedException.getCause())
612-
.hasMessageThat()
613-
.contains("Response contained invalid X-Goog-Upload-Size-Received header: not-a-number");
626+
assertThat(Throwables.getRootCause(malformedException))
627+
.isInstanceOf(NumberFormatException.class);
614628
}
615629

616630
private static HttpJsonResumableUploadClient<TestRequest, String> createClient(

0 commit comments

Comments
 (0)