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,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;
@@ -59,7 +60,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 +165,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 +178,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 +194,15 @@ 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 if (uploadStatus == ResumableUploadStatus.FINAL) {
+          future.setException(createServerRejectionException(statusCode, uploadStatus, trailers));
         } else {
           Throwable cause = trailers.getException();
           future.setException(
@@ -225,4 +216,31 @@ public void onClose(int statusCode, HttpJsonMetadata trailers) {
       }
     }
   }
+
+  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) {
+    StatusCode.Code canonicalCode = HttpJsonStatusCode.of(httpStatusCode).getCode();
+    return new StatusCode() {
+      @Override
+      public Code getCode() {
+        return canonicalCode;
+      }
+
+      @Override
+      public @Nullable Integer getTransportCode() {
+        // null transportCode ensures error is classified as FATAL
+        return null;
+      }
+    };
+  }
 }

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(
@@ -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(

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

@@ -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;
@@ -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");
@@ -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");
@@ -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");
@@ -379,7 +381,7 @@ void uploadChunk_serverReturnsConflictOrError_throwsException() {
   }
 
   @Test
-  void uploadChunk_missingUploadStatusHeader_throwsInternalException() {
+  void uploadChunk_missingUploadStatusHeader_returnsUnknownUploadStatusOnHttp200() {
     MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse();
     httpResponse.setStatusCode(200);
 
@@ -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
@@ -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
@@ -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");
   }
@@ -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

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