Skip to content

Commit 4ee20f2

Browse files
whoweswhowes
committed
feat(gax): add upload-status header plumbing
Add nullable getUploadStatus() accessors to ChunkUploadResponse, QueryStatusResponse, and ResumableUploadSession, and plumb the X-Goog-Upload-Status response header through the HTTP/JSON callables. On HTTP 200 chunk responses where X-Goog-Upload-Status is absent, return a ChunkUploadResponse with a null uploadStatus rather than throwing a wire-level exception, allowing higher-level upload coordinators to classify the missing header and trigger protocol recovery. Existing test uploadChunk_missingUploadStatusHeader_throwsInternalException was updated to uploadChunk_missingUploadStatusHeader_returnsNullUploadStatusOnHttp200 to reflect that the missing status header on HTTP 200 is now surfaced via a null status property on ChunkUploadResponse instead of throwing an InternalException at the transport layer. Note: No end-to-end integration test is included because the test server always returns the X-Goog-Upload-Status header on success, making header absence uninjectable end-to-end.
1 parent 2419e04 commit 4ee20f2

8 files changed

Lines changed: 223 additions & 81 deletions

File tree

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

Lines changed: 39 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
import com.google.api.core.ApiFuture;
3434
import com.google.api.gax.resumable.ChunkUploadRequest;
3535
import com.google.api.gax.resumable.ChunkUploadResponse;
36+
import com.google.api.gax.resumable.ResumableUploadStatus;
3637
import com.google.api.gax.rpc.ApiCallContext;
3738
import com.google.api.gax.rpc.ApiExceptionFactory;
3839
import com.google.api.gax.rpc.ClientContext;
@@ -59,7 +60,6 @@ class ResumableUploadChunkCallable<ResponseT>
5960
private static final String UPLOAD_COMMAND_HEADER = "X-Goog-Upload-Command";
6061
private static final String UPLOAD_OFFSET_HEADER = "X-Goog-Upload-Offset";
6162
private static final String UPLOAD_STATUS_HEADER = "X-Goog-Upload-Status";
62-
private static final String STATUS_FINAL = "final";
6363

6464
private static final String COMMAND_UPLOAD = "upload";
6565
private static final String COMMAND_FINALIZE = "finalize";
@@ -165,7 +165,7 @@ private static class ChunkUploadResponseListener<ResponseT>
165165

166166
private final ResumableUploadHttpJsonFuture<ChunkUploadResponse<ResponseT>> future;
167167
private final HttpResponseParser<ResponseT> responseParser;
168-
@Nullable private String uploadStatus = null;
168+
private ResumableUploadStatus uploadStatus = ResumableUploadStatus.UNKNOWN;
169169
private String responseBody = "";
170170

171171
private ChunkUploadResponseListener(
@@ -178,7 +178,9 @@ private ChunkUploadResponseListener(
178178
@Override
179179
public void onHeaders(HttpJsonMetadata responseHeaders) {
180180
Map<String, Object> headers = responseHeaders.getHeaders();
181-
this.uploadStatus = HttpHeadersUtils.getSingleHeader(headers, UPLOAD_STATUS_HEADER);
181+
this.uploadStatus =
182+
ResumableUploadStatus.fromHeader(
183+
HttpHeadersUtils.getSingleHeader(headers, UPLOAD_STATUS_HEADER));
182184
}
183185

184186
@Override
@@ -191,27 +193,19 @@ public void onMessage(@Nullable String message) {
191193
@Override
192194
public void onClose(int statusCode, HttpJsonMetadata trailers) {
193195
try {
194-
if (statusCode >= 200 && statusCode < 300) {
195-
if (uploadStatus == null) {
196-
future.setException(
197-
ApiExceptionFactory.createException(
198-
"Upload chunk response did not contain valid "
199-
+ UPLOAD_STATUS_HEADER
200-
+ " header",
201-
/* cause= */ null,
202-
HttpJsonStatusCode.of(StatusCode.Code.INTERNAL),
203-
/* retryable= */ false));
204-
return;
205-
}
206-
boolean isComplete = STATUS_FINAL.equalsIgnoreCase(uploadStatus);
207-
ChunkUploadResponse.Builder<ResponseT> chunkResponseBuilder =
208-
ChunkUploadResponse.<ResponseT>newBuilder().setComplete(isComplete);
209-
if (isComplete) {
196+
if (statusCode >= 200
197+
&& statusCode < 300
198+
&& uploadStatus != ResumableUploadStatus.CANCELLED) {
199+
ResponseT response = null;
200+
if (uploadStatus == ResumableUploadStatus.FINAL) {
210201
InputStream stream =
211202
new ByteArrayInputStream(responseBody.getBytes(StandardCharsets.UTF_8));
212-
chunkResponseBuilder.setResponse(responseParser.parse(stream));
203+
response = responseParser.parse(stream);
213204
}
214-
future.set(chunkResponseBuilder.build());
205+
future.set(ChunkUploadResponse.create(uploadStatus, response));
206+
} else if (uploadStatus == ResumableUploadStatus.CANCELLED
207+
|| uploadStatus == ResumableUploadStatus.FINAL) {
208+
future.setException(createServerRejectionException(statusCode, uploadStatus, trailers));
215209
} else {
216210
Throwable cause = trailers.getException();
217211
future.setException(
@@ -225,4 +219,28 @@ public void onClose(int statusCode, HttpJsonMetadata trailers) {
225219
}
226220
}
227221
}
222+
223+
private static final StatusCode FAILED_PRECONDITION_STATUS_CODE =
224+
new StatusCode() {
225+
@Override
226+
public Code getCode() {
227+
return Code.FAILED_PRECONDITION;
228+
}
229+
230+
@Override
231+
public @Nullable Integer getTransportCode() {
232+
return null;
233+
}
234+
};
235+
236+
static Throwable createServerRejectionException(
237+
int statusCode, ResumableUploadStatus uploadStatus, HttpJsonMetadata trailers) {
238+
Throwable cause = trailers.getException();
239+
String message = "Upload " + uploadStatus + " by server with status code: " + statusCode;
240+
if (cause != null && cause.getMessage() != null) {
241+
message = cause.getMessage();
242+
}
243+
return ApiExceptionFactory.createException(
244+
message, cause, FAILED_PRECONDITION_STATUS_CODE, /* retryable= */ false);
245+
}
228246
}

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

Lines changed: 20 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
import com.google.api.core.ApiFuture;
3434
import com.google.api.gax.resumable.QueryStatusRequest;
3535
import com.google.api.gax.resumable.QueryStatusResponse;
36+
import com.google.api.gax.resumable.ResumableUploadStatus;
3637
import com.google.api.gax.rpc.ApiCallContext;
3738
import com.google.api.gax.rpc.ApiExceptionFactory;
3839
import com.google.api.gax.rpc.ClientContext;
@@ -63,7 +64,6 @@ class ResumableUploadQueryStatusCallable<ResponseT>
6364
private static final String UPLOAD_COMMAND_HEADER = "X-Goog-Upload-Command";
6465
private static final String UPLOAD_STATUS_HEADER = "X-Goog-Upload-Status";
6566
private static final String UPLOAD_SIZE_RECEIVED_HEADER = "X-Goog-Upload-Size-Received";
66-
private static final String STATUS_FINAL = "final";
6767
private static final String COMMAND_QUERY = "query";
6868

6969
private static final Map<String, List<String>> QUERY_STATUS_HEADERS =
@@ -182,7 +182,7 @@ private static class QueryStatusResponseListener<ResponseT>
182182

183183
private final ResumableUploadHttpJsonFuture<QueryStatusResponse<ResponseT>> future;
184184
private final HttpResponseParser<ResponseT> responseParser;
185-
@Nullable private String uploadStatus = null;
185+
private ResumableUploadStatus uploadStatus = ResumableUploadStatus.UNKNOWN;
186186
@Nullable private Long committedOffset = null;
187187
@Nullable private Throwable headerParsingException;
188188
private String responseBody = "";
@@ -197,7 +197,9 @@ private QueryStatusResponseListener(
197197
@Override
198198
public void onHeaders(HttpJsonMetadata responseHeaders) {
199199
Map<String, Object> headers = responseHeaders.getHeaders();
200-
this.uploadStatus = HttpHeadersUtils.getSingleHeader(headers, UPLOAD_STATUS_HEADER);
200+
this.uploadStatus =
201+
ResumableUploadStatus.fromHeader(
202+
HttpHeadersUtils.getSingleHeader(headers, UPLOAD_STATUS_HEADER));
201203
try {
202204
this.committedOffset = parseSizeReceived(responseHeaders);
203205
} catch (Throwable t) {
@@ -215,24 +217,26 @@ public void onMessage(@Nullable String message) {
215217
@Override
216218
public void onClose(int statusCode, HttpJsonMetadata trailers) {
217219
try {
218-
if (statusCode >= 200 && statusCode < 300) {
220+
if (statusCode >= 200
221+
&& statusCode < 300
222+
&& uploadStatus != ResumableUploadStatus.CANCELLED) {
219223
if (headerParsingException != null) {
220224
future.setException(headerParsingException);
221225
return;
222226
}
223-
boolean isComplete = STATUS_FINAL.equalsIgnoreCase(uploadStatus);
224-
if (isComplete) {
225-
QueryStatusResponse.Builder<ResponseT> queryResponseBuilder =
226-
QueryStatusResponse.<ResponseT>newBuilder().setComplete(true);
227+
if (uploadStatus == ResumableUploadStatus.FINAL) {
227228
InputStream stream =
228229
new ByteArrayInputStream(responseBody.getBytes(StandardCharsets.UTF_8));
229-
queryResponseBuilder.setResponse(responseParser.parse(stream));
230-
future.set(queryResponseBuilder.build());
230+
future.set(
231+
QueryStatusResponse.<ResponseT>newBuilder()
232+
.setUploadStatus(uploadStatus)
233+
.setResponse(responseParser.parse(stream))
234+
.build());
231235
} else if (committedOffset != null) {
232236
future.set(
233237
QueryStatusResponse.<ResponseT>newBuilder()
234-
.setComplete(false)
235238
.setCommittedOffset(committedOffset)
239+
.setUploadStatus(uploadStatus)
236240
.build());
237241
} else {
238242
future.setException(
@@ -244,6 +248,11 @@ public void onClose(int statusCode, HttpJsonMetadata trailers) {
244248
HttpJsonStatusCode.of(StatusCode.Code.INTERNAL),
245249
/* retryable= */ false));
246250
}
251+
} else if (uploadStatus == ResumableUploadStatus.CANCELLED
252+
|| uploadStatus == ResumableUploadStatus.FINAL) {
253+
future.setException(
254+
ResumableUploadChunkCallable.createServerRejectionException(
255+
statusCode, uploadStatus, trailers));
247256
} else {
248257
Throwable cause = trailers.getException();
249258
future.setException(

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

Lines changed: 42 additions & 19 deletions
Original file line numberDiff line numberDiff line change
@@ -45,10 +45,11 @@
4545
import com.google.api.gax.resumable.QueryStatusRequest;
4646
import com.google.api.gax.resumable.QueryStatusResponse;
4747
import com.google.api.gax.resumable.ResumableUploadSession;
48+
import com.google.api.gax.resumable.ResumableUploadStatus;
4849
import com.google.api.gax.rpc.AbortedException;
4950
import com.google.api.gax.rpc.ApiCallContext;
50-
import com.google.api.gax.rpc.ApiException;
5151
import com.google.api.gax.rpc.ClientContext;
52+
import com.google.api.gax.rpc.FailedPreconditionException;
5253
import com.google.api.gax.rpc.InternalException;
5354
import com.google.api.gax.rpc.NotFoundException;
5455
import com.google.api.gax.rpc.StatusCode;
@@ -260,8 +261,8 @@ void uploadChunk_intermediateChunk_sendsUploadCommandAndReturnsActiveStatus() {
260261

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

263-
assertThat(response.isComplete()).isFalse();
264264
assertThat(response.getResponse()).isNull();
265+
assertThat(response.getUploadStatus()).isEqualTo(ResumableUploadStatus.ACTIVE);
265266

266267
assertThat(transport.capturedUrl).isEqualTo(TEST_UPLOAD_URL);
267268
assertThat(transport.capturedHeaders.get("x-goog-upload-command")).containsExactly("upload");
@@ -289,9 +290,9 @@ void uploadChunk_finalChunk_sendsUploadFinalizeAndReturnsResponseBody() {
289290

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

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

296297
assertThat(transport.capturedHeaders.get("x-goog-upload-command"))
297298
.containsExactly("upload, finalize");
@@ -318,9 +319,9 @@ void uploadChunk_emptyPayloadFinal_sendsFinalizeCommandAndReturnsResponseBody()
318319

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

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

325326
assertThat(transport.capturedHeaders.get("x-goog-upload-command")).containsExactly("finalize");
326327
assertThat(transport.capturedHeaders).doesNotContainKey("x-goog-upload-offset");
@@ -379,7 +380,7 @@ void uploadChunk_serverReturnsConflictOrError_throwsException() {
379380
}
380381

381382
@Test
382-
void uploadChunk_missingUploadStatusHeader_throwsInternalException() {
383+
void uploadChunk_missingUploadStatusHeader_returnsUnknownUploadStatusOnHttp200() {
383384
MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse();
384385
httpResponse.setStatusCode(200);
385386

@@ -391,18 +392,14 @@ void uploadChunk_missingUploadStatusHeader_throwsInternalException() {
391392
.setOffset(0L)
392393
.build();
393394

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

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

404401
@Test
405-
void uploadChunk_serverReturnsFinalStatusOnNon200_marksExceptionNonRetryable() {
402+
void uploadChunk_serverReturnsFinalStatusOnNon200_throwsFailedPreconditionException() {
406403
MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse();
407404
httpResponse.setStatusCode(503);
408405
httpResponse.addHeader("X-Goog-Upload-Status", "final");
@@ -420,10 +417,36 @@ void uploadChunk_serverReturnsFinalStatusOnNon200_marksExceptionNonRetryable() {
420417
assertThrows(
421418
ExecutionException.class, () -> client.uploadChunkCallable().futureCall(request).get());
422419

423-
assertThat(exception.getCause()).isInstanceOf(ApiException.class);
424-
ApiException apiException = (ApiException) exception.getCause();
425-
assertThat(apiException.isRetryable()).isFalse();
426-
assertThat(apiException.getStatusCode().getCode()).isEqualTo(StatusCode.Code.UNAVAILABLE);
420+
assertThat(exception.getCause()).isInstanceOf(FailedPreconditionException.class);
421+
FailedPreconditionException failedPrecondition =
422+
(FailedPreconditionException) exception.getCause();
423+
assertThat(failedPrecondition.getStatusCode().getCode())
424+
.isEqualTo(StatusCode.Code.FAILED_PRECONDITION);
425+
}
426+
427+
@Test
428+
void uploadChunk_serverReturnsCancelledStatus_throwsFailedPreconditionException() {
429+
MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse();
430+
httpResponse.setStatusCode(200);
431+
httpResponse.addHeader("X-Goog-Upload-Status", "cancelled");
432+
433+
HttpJsonResumableUploadClient<TestRequest, String> client = createClient(httpResponse);
434+
ChunkUploadRequest request =
435+
ChunkUploadRequest.newBuilder()
436+
.setUploadUrl(TEST_UPLOAD_URL)
437+
.setPayload("data".getBytes(StandardCharsets.UTF_8))
438+
.setOffset(0L)
439+
.build();
440+
441+
ExecutionException exception =
442+
assertThrows(
443+
ExecutionException.class, () -> client.uploadChunkCallable().futureCall(request).get());
444+
445+
assertThat(exception.getCause()).isInstanceOf(FailedPreconditionException.class);
446+
FailedPreconditionException failedPrecondition =
447+
(FailedPreconditionException) exception.getCause();
448+
assertThat(failedPrecondition.getStatusCode().getCode())
449+
.isEqualTo(StatusCode.Code.FAILED_PRECONDITION);
427450
}
428451

429452
@Test
@@ -439,9 +462,9 @@ void queryStatus_activeUpload_returnsCommittedOffset() {
439462

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

442-
assertThat(response.isComplete()).isFalse();
443465
assertThat(response.getCommittedOffset()).isEqualTo(524288L);
444466
assertThat(response.getResponse()).isNull();
467+
assertThat(response.getUploadStatus()).isEqualTo(ResumableUploadStatus.ACTIVE);
445468

446469
assertThat(transport.capturedHeaders.get("x-goog-upload-command")).containsExactly("query");
447470
}
@@ -458,10 +481,10 @@ void queryStatus_finalUpload_returnsCompleteAndResponseBody() {
458481

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

461-
assertThat(response.isComplete()).isTrue();
462484
assertThat(response.getCommittedOffset()).isNull();
463485
assertThat(response.getResponse())
464486
.isEqualTo("{\"name\":\"uploaded-file.txt\",\"size\":1048576}");
487+
assertThat(response.getUploadStatus()).isEqualTo(ResumableUploadStatus.FINAL);
465488
}
466489

467490
@Test

‎sdk-platform-java/gax-java/gax/src/main/java/com/google/api/gax/resumable/ChunkUploadResponse.java‎

Lines changed: 9 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -46,35 +46,36 @@
4646
@AutoValue
4747
public abstract class ChunkUploadResponse<ResponseT> {
4848

49-
/** Whether the overall resumable upload stream has finalized and completed on the server. */
50-
public abstract boolean isComplete();
51-
5249
/**
5350
* The response object returned by the server upon final completion (e.g. metadata of the uploaded
5451
* resource), or {@code null} if the upload is still in progress.
5552
*/
5653
public abstract @Nullable ResponseT getResponse();
5754

55+
/** Returns the status of the upload session returned by the server. */
56+
public abstract ResumableUploadStatus getUploadStatus();
57+
5858
public abstract Builder<ResponseT> toBuilder();
5959

6060
public static <ResponseT> Builder<ResponseT> newBuilder() {
61-
return new AutoValue_ChunkUploadResponse.Builder<ResponseT>().setComplete(false);
61+
return new AutoValue_ChunkUploadResponse.Builder<ResponseT>()
62+
.setUploadStatus(ResumableUploadStatus.ACTIVE);
6263
}
6364

6465
public static <ResponseT> ChunkUploadResponse<ResponseT> create(
65-
boolean isComplete, @Nullable ResponseT response) {
66+
ResumableUploadStatus uploadStatus, @Nullable ResponseT response) {
6667
return new AutoValue_ChunkUploadResponse.Builder<ResponseT>()
67-
.setComplete(isComplete)
68+
.setUploadStatus(uploadStatus)
6869
.setResponse(response)
6970
.build();
7071
}
7172

7273
@AutoValue.Builder
7374
public abstract static class Builder<ResponseT> {
74-
public abstract Builder<ResponseT> setComplete(boolean isComplete);
75-
7675
public abstract Builder<ResponseT> setResponse(@Nullable ResponseT response);
7776

77+
public abstract Builder<ResponseT> setUploadStatus(ResumableUploadStatus uploadStatus);
78+
7879
public abstract ChunkUploadResponse<ResponseT> build();
7980
}
8081
}

0 commit comments

Comments
 (0)