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.

Author: whowes

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

@@ -33,10 +33,9 @@
 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;
-import com.google.api.gax.rpc.StatusCode;
 import com.google.api.gax.rpc.UnaryCallable;
 import com.google.api.pathtemplate.PathTemplate;
 import com.google.common.base.Preconditions;
@@ -59,7 +58,6 @@ class ResumableUploadChunkCallable<ResponseT>
   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";
@@ -165,7 +163,7 @@ private static class ChunkUploadResponseListener<ResponseT>
 
     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(
@@ -178,7 +176,9 @@ private ChunkUploadResponseListener(
     @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
@@ -192,26 +192,13 @@ public void onMessage(@Nullable String message) {
     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 {
           Throwable cause = trailers.getException();
           future.setException(

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

@@ -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;
@@ -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 =
@@ -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 = "";
@@ -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) {
@@ -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(

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

@@ -45,9 +45,9 @@
 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.NotFoundException;
@@ -260,8 +260,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");
@@ -289,9 +289,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");
@@ -318,9 +318,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");
@@ -379,7 +379,7 @@ void uploadChunk_serverReturnsConflictOrError_throwsException() {
   }
 
   @Test
-  void uploadChunk_missingUploadStatusHeader_throwsInternalException() {
+  void uploadChunk_missingUploadStatusHeader_returnsUnknownUploadStatusOnHttp200() {
     MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse();
     httpResponse.setStatusCode(200);
 
@@ -391,39 +391,10 @@ void uploadChunk_missingUploadStatusHeader_throwsInternalException() {
             .setOffset(0L)
             .build();
 
-    ExecutionException exception =
-        assertThrows(
-            ExecutionException.class, () -> client.uploadChunkCallable().futureCall(request).get());
-
-    assertThat(exception.getCause()).isInstanceOf(InternalException.class);
-    assertThat(exception.getCause())
-        .hasMessageThat()
-        .contains("Upload chunk response did not contain valid X-Goog-Upload-Status header");
-  }
-
-  @Test
-  void uploadChunk_serverReturnsFinalStatusOnNon200_marksExceptionNonRetryable() {
-    MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse();
-    httpResponse.setStatusCode(503);
-    httpResponse.addHeader("X-Goog-Upload-Status", "final");
-    httpResponse.setContent("{\"error\":{\"message\":\"Upload rejected by backend\"}}");
-
-    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());
+    ChunkUploadResponse<String> response = client.uploadChunkCallable().call(request);
 
-    assertThat(exception.getCause()).isInstanceOf(ApiException.class);
-    ApiException apiException = (ApiException) exception.getCause();
-    assertThat(apiException.isRetryable()).isFalse();
-    assertThat(apiException.getStatusCode().getCode()).isEqualTo(StatusCode.Code.UNAVAILABLE);
+    assertThat(response.getResponse()).isNull();
+    assertThat(response.getUploadStatus()).isEqualTo(ResumableUploadStatus.UNKNOWN);
   }
 
   @Test
@@ -439,9 +410,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");
   }
@@ -458,10 +429,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

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

@@ -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();
   }
 }

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

@@ -51,34 +51,35 @@ public abstract class QueryStatusResponse<ResponseT> {
    * null} if the server did not return a committed offset (e.g. if the upload is already
    * finalized).
    *
-   * <p>When {@link #isComplete()} is {@code false}, this value is guaranteed to be non-null and
-   * represents the starting offset for resuming the upload.
+   * <p>When {@link #getUploadStatus()} is {@link ResumableUploadStatus#ACTIVE}, this value is
+   * guaranteed to be non-null and represents the starting offset for resuming the upload.
    */
   public abstract @Nullable Long getCommittedOffset();
 
-  /** Whether the resumable upload session 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_QueryStatusResponse.Builder<ResponseT>().setComplete(false);
+    return new AutoValue_QueryStatusResponse.Builder<ResponseT>()
+        .setUploadStatus(ResumableUploadStatus.ACTIVE);
   }
 
   @AutoValue.Builder
   public abstract static class Builder<ResponseT> {
     public abstract Builder<ResponseT> setCommittedOffset(@Nullable Long committedOffset);
 
-    public abstract Builder<ResponseT> setComplete(boolean isComplete);
-
     public abstract Builder<ResponseT> setResponse(@Nullable ResponseT response);
 
+    public abstract Builder<ResponseT> setUploadStatus(ResumableUploadStatus uploadStatus);
+
     public abstract QueryStatusResponse<ResponseT> build();
   }
 }