feat(gax): add upload-status header plumbing

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

@@ -51,6 +51,10 @@ public static HttpJsonStatusCode of(Code statusCode) {
     return new HttpJsonStatusCode(statusCode.getHttpStatusCode(), statusCode);
   }
 
+  static HttpJsonStatusCode of(int httpStatus, Code statusCode) {
+    return new HttpJsonStatusCode(httpStatus, statusCode);
+  }
+
   public static HttpJsonStatusCode of(com.google.rpc.Code rpcCode) {
     return HttpJsonStatusCode.of(rpcCodeToStatusCode(rpcCode));
   }

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

@@ -60,6 +60,7 @@
   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 STATUS_CANCELLED = "cancelled";
 
   private static final String COMMAND_UPLOAD = "upload";
   private static final String COMMAND_FINALIZE = "finalize";
@@ -189,29 +190,44 @@
     }
 
     @Override
     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;
-          }
+        if (STATUS_CANCELLED.equalsIgnoreCase(uploadStatus)) {
+          Throwable cause = trailers.getException();
+          String message =
+              cause != null && cause.getMessage() != null
+                  ? cause.getMessage()
+                  : "Upload cancelled by server with status code: " + statusCode;
+          future.setException(
+              ApiExceptionFactory.createException(
+                  message,
+                  cause,
+                  HttpJsonStatusCode.of(statusCode, StatusCode.Code.FAILED_PRECONDITION),
+                  false));
+        } else if (statusCode >= 200 && statusCode < 300) {
           boolean isComplete = STATUS_FINAL.equalsIgnoreCase(uploadStatus);
           ChunkUploadResponse.Builder<ResponseT> chunkResponseBuilder =
-              ChunkUploadResponse.<ResponseT>newBuilder().setComplete(isComplete);
+              ChunkUploadResponse.<ResponseT>newBuilder()
+                  .setComplete(isComplete)
+                  .setUploadStatus(uploadStatus);
           if (isComplete) {
             InputStream stream =
                 new ByteArrayInputStream(responseBody.getBytes(StandardCharsets.UTF_8));
             chunkResponseBuilder.setResponse(responseParser.parse(stream));
           }
           future.set(chunkResponseBuilder.build());
+        } else if (STATUS_FINAL.equalsIgnoreCase(uploadStatus)) {
+          Throwable cause = trailers.getException();
+          String message =
+              cause != null && cause.getMessage() != null
+                  ? cause.getMessage()
+                  : "Upload rejected by server with status code: " + statusCode;
+          future.setException(
+              ApiExceptionFactory.createException(
+                  message,
+                  cause,
+                  HttpJsonStatusCode.of(statusCode, StatusCode.Code.FAILED_PRECONDITION),
+                  false));
         } else {
           Throwable cause = trailers.getException();
           future.setException(

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

@@ -64,6 +64,7 @@
   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 STATUS_CANCELLED = "cancelled";
   private static final String COMMAND_QUERY = "query";
 
   private static final Map<String, List<String>> QUERY_STATUS_HEADERS =
@@ -213,17 +214,31 @@
     }
 
     @Override
     public void onClose(int statusCode, HttpJsonMetadata trailers) {
       try {
-        if (statusCode >= 200 && statusCode < 300) {
+        if (STATUS_CANCELLED.equalsIgnoreCase(uploadStatus)) {
+          Throwable cause = trailers.getException();
+          String message =
+              cause != null && cause.getMessage() != null
+                  ? cause.getMessage()
+                  : "Upload cancelled by server with status code: " + statusCode;
+          future.setException(
+              ApiExceptionFactory.createException(
+                  message,
+                  cause,
+                  HttpJsonStatusCode.of(statusCode, StatusCode.Code.FAILED_PRECONDITION),
+                  false));
+        } else if (statusCode >= 200 && statusCode < 300) {
           if (headerParsingException != null) {
             future.setException(headerParsingException);
             return;
           }
           boolean isComplete = STATUS_FINAL.equalsIgnoreCase(uploadStatus);
           if (isComplete) {
             QueryStatusResponse.Builder<ResponseT> queryResponseBuilder =
-                QueryStatusResponse.<ResponseT>newBuilder().setComplete(true);
+                QueryStatusResponse.<ResponseT>newBuilder()
+                    .setComplete(true)
+                    .setUploadStatus(uploadStatus);
             InputStream stream =
                 new ByteArrayInputStream(responseBody.getBytes(StandardCharsets.UTF_8));
             queryResponseBuilder.setResponse(responseParser.parse(stream));
@@ -233,6 +248,7 @@
                 QueryStatusResponse.<ResponseT>newBuilder()
                     .setComplete(false)
                     .setCommittedOffset(committedOffset)
+                    .setUploadStatus(uploadStatus)
                     .build());
           } else {
             future.setException(
@@ -244,6 +260,18 @@
                     HttpJsonStatusCode.of(StatusCode.Code.INTERNAL),
                     /* retryable= */ false));
           }
+        } else if (STATUS_FINAL.equalsIgnoreCase(uploadStatus)) {
+          Throwable cause = trailers.getException();
+          String message =
+              cause != null && cause.getMessage() != null
+                  ? cause.getMessage()
+                  : "Upload rejected by server with status code: " + statusCode;
+          future.setException(
+              ApiExceptionFactory.createException(
+                  message,
+                  cause,
+                  HttpJsonStatusCode.of(statusCode, StatusCode.Code.FAILED_PRECONDITION),
+                  false));
         } else {
           Throwable cause = trailers.getException();
           future.setException(

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

@@ -55,6 +55,7 @@ class ResumableUploadStartCallable<RequestT>
   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 String UPLOAD_STATUS_HEADER = "X-Goog-Upload-Status";
 
   private static final Map<String, List<String>> START_UPLOAD_HEADERS =
       ImmutableMap.of(
@@ -109,6 +110,7 @@ private static class StartUploadResponseListener extends HttpJsonClientCall.List
     private final ResumableUploadHttpJsonFuture<ResumableUploadSession> future;
     private long chunkGranularity = 1L;
     @Nullable private String uploadUrl;
+    @Nullable private String uploadStatus;
     @Nullable private Throwable headerParsingException;
 
     private StartUploadResponseListener(
@@ -120,6 +122,8 @@ private StartUploadResponseListener(
     public void onHeaders(HttpJsonMetadata responseHeaders) {
       Map<String, Object> headers = responseHeaders.getHeaders();
 
+      this.uploadStatus = HttpHeadersUtils.getSingleHeader(headers, UPLOAD_STATUS_HEADER);
+
       String url = HttpHeadersUtils.getSingleHeader(headers, UPLOAD_URL_HEADER);
       if (!Strings.isNullOrEmpty(url)) {
         this.uploadUrl = url;
@@ -172,6 +176,7 @@ public void onClose(int statusCode, HttpJsonMetadata trailers) {
                 ResumableUploadSession.newBuilder()
                     .setUploadUrl(uploadUrl)
                     .setChunkGranularity(chunkGranularity)
+                    .setUploadStatus(uploadStatus)
                     .build());
           } else {
             future.setException(

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

@@ -47,8 +47,8 @@
 import com.google.api.gax.resumable.ResumableUploadSession;
 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.FailedPreconditionException;
 import com.google.api.gax.rpc.InternalException;
 import com.google.api.gax.rpc.NotFoundException;
 import com.google.api.gax.rpc.StatusCode;
@@ -100,6 +100,23 @@ void startUpload_validHeaders_returnsSession() {
 
     assertThat(session.getUploadUrl()).isEqualTo(TEST_UPLOAD_URL);
     assertThat(session.getChunkGranularity()).isEqualTo(262144L);
+    assertThat(session.getUploadStatus()).isNull();
+  }
+
+  @Test
+  void startUpload_withUploadStatusHeader_returnsSessionWithUploadStatus() {
+    MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse();
+    httpResponse.setStatusCode(200);
+    httpResponse.addHeader("X-Goog-Upload-URL", TEST_UPLOAD_URL);
+    httpResponse.addHeader("X-Goog-Upload-Status", "active");
+
+    HttpJsonResumableUploadClient<TestRequest, String> client = createClient(httpResponse);
+    TestRequest request = new TestRequest("upload/v1/resources");
+
+    ResumableUploadSession session = client.startUploadCallable().call(request);
+
+    assertThat(session.getUploadUrl()).isEqualTo(TEST_UPLOAD_URL);
+    assertThat(session.getUploadStatus()).isEqualTo("active");
   }
 
   @Test
@@ -262,6 +279,7 @@ void uploadChunk_intermediateChunk_sendsUploadCommandAndReturnsActiveStatus() {
 
     assertThat(response.isComplete()).isFalse();
     assertThat(response.getResponse()).isNull();
+    assertThat(response.getUploadStatus()).isEqualTo("active");
 
     assertThat(transport.capturedUrl).isEqualTo(TEST_UPLOAD_URL);
     assertThat(transport.capturedHeaders.get("x-goog-upload-command")).containsExactly("upload");
@@ -292,6 +310,7 @@ void uploadChunk_finalChunk_sendsUploadFinalizeAndReturnsResponseBody() {
     assertThat(response.isComplete()).isTrue();
     assertThat(response.getResponse())
         .isEqualTo("{\"name\":\"uploaded-file.txt\",\"size\":524288}");
+    assertThat(response.getUploadStatus()).isEqualTo("final");
 
     assertThat(transport.capturedHeaders.get("x-goog-upload-command"))
         .containsExactly("upload, finalize");
@@ -321,6 +340,7 @@ void uploadChunk_emptyPayloadFinal_sendsFinalizeCommandAndReturnsResponseBody()
     assertThat(response.isComplete()).isTrue();
     assertThat(response.getResponse())
         .isEqualTo("{\"name\":\"uploaded-file.txt\",\"size\":1048576}");
+    assertThat(response.getUploadStatus()).isEqualTo("final");
 
     assertThat(transport.capturedHeaders.get("x-goog-upload-command")).containsExactly("finalize");
     assertThat(transport.capturedHeaders).doesNotContainKey("x-goog-upload-offset");
@@ -379,10 +399,32 @@ void uploadChunk_serverReturnsConflictOrError_throwsException() {
   }
 
   @Test
-  void uploadChunk_missingUploadStatusHeader_throwsInternalException() {
+  void uploadChunk_missingUploadStatusHeader_returnsNullUploadStatusOnHttp200() {
     MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse();
     httpResponse.setStatusCode(200);
 
+    HttpJsonResumableUploadClient<TestRequest, String> client = createClient(httpResponse);
+    ChunkUploadRequest request =
+        ChunkUploadRequest.newBuilder()
+            .setUploadUrl(TEST_UPLOAD_URL)
+            .setPayload("data".getBytes(StandardCharsets.UTF_8))
+            .setOffset(0L)
+            .build();
+
+    ChunkUploadResponse<String> response = client.uploadChunkCallable().call(request);
+
+    assertThat(response.isComplete()).isFalse();
+    assertThat(response.getResponse()).isNull();
+    assertThat(response.getUploadStatus()).isNull();
+  }
+
+  @Test
+  void uploadChunk_serverReturnsFinalStatusOnNon200_throwsFailedPreconditionException() {
+    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()
@@ -395,18 +437,19 @@ void uploadChunk_missingUploadStatusHeader_throwsInternalException() {
         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");
+    assertThat(exception.getCause()).isInstanceOf(FailedPreconditionException.class);
+    FailedPreconditionException failedPrecondition =
+        (FailedPreconditionException) exception.getCause();
+    assertThat(failedPrecondition.getStatusCode().getCode())
+        .isEqualTo(StatusCode.Code.FAILED_PRECONDITION);
   }
 
   @Test
-  void uploadChunk_serverReturnsFinalStatusOnNon200_marksExceptionNonRetryable() {
+  void uploadChunk_serverReturnsCancelledStatus_throwsFailedPreconditionException() {
     MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse();
     httpResponse.setStatusCode(503);
-    httpResponse.addHeader("X-Goog-Upload-Status", "final");
-    httpResponse.setContent("{\"error\":{\"message\":\"Upload rejected by backend\"}}");
+    httpResponse.addHeader("X-Goog-Upload-Status", "cancelled");
+    httpResponse.setContent("{\"error\":{\"message\":\"Upload cancelled by backend\"}}");
 
     HttpJsonResumableUploadClient<TestRequest, String> client = createClient(httpResponse);
     ChunkUploadRequest request =
@@ -420,10 +463,11 @@ 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(FailedPreconditionException.class);
+    FailedPreconditionException failedPrecondition =
+        (FailedPreconditionException) exception.getCause();
+    assertThat(failedPrecondition.getStatusCode().getCode())
+        .isEqualTo(StatusCode.Code.FAILED_PRECONDITION);
   }
 
   @Test
@@ -442,6 +486,7 @@ void queryStatus_activeUpload_returnsCommittedOffset() {
     assertThat(response.isComplete()).isFalse();
     assertThat(response.getCommittedOffset()).isEqualTo(524288L);
     assertThat(response.getResponse()).isNull();
+    assertThat(response.getUploadStatus()).isEqualTo("active");
 
     assertThat(transport.capturedHeaders.get("x-goog-upload-command")).containsExactly("query");
   }
@@ -462,6 +507,23 @@ void queryStatus_finalUpload_returnsCompleteAndResponseBody() {
     assertThat(response.getCommittedOffset()).isNull();
     assertThat(response.getResponse())
         .isEqualTo("{\"name\":\"uploaded-file.txt\",\"size\":1048576}");
+    assertThat(response.getUploadStatus()).isEqualTo("final");
+  }
+
+  @Test
+  void queryStatus_missingUploadStatusHeader_returnsNullUploadStatusOnHttp200() {
+    MockLowLevelHttpResponse httpResponse = new MockLowLevelHttpResponse();
+    httpResponse.setStatusCode(200);
+    httpResponse.addHeader("X-Goog-Upload-Size-Received", "524288");
+
+    HttpJsonResumableUploadClient<TestRequest, String> client = createClient(httpResponse);
+    QueryStatusRequest request = QueryStatusRequest.create(TEST_UPLOAD_URL);
+
+    QueryStatusResponse<String> response = client.queryStatusCallable().call(request);
+
+    assertThat(response.isComplete()).isFalse();
+    assertThat(response.getCommittedOffset()).isEqualTo(524288L);
+    assertThat(response.getUploadStatus()).isNull();
   }
 
   @Test

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

@@ -55,6 +55,12 @@
    */
   public abstract @Nullable ResponseT getResponse();
 
+  /**
+   * Returns the status of the upload session returned by the server (e.g. {@code "active"} or
+   * {@code "final"}), or {@code null} if the server omitted the upload status header.
+   */
+  public abstract @Nullable String getUploadStatus();
+
   public abstract Builder<ResponseT> toBuilder();
 
   public static <ResponseT> Builder<ResponseT> newBuilder() {
@@ -63,9 +69,15 @@
 
   public static <ResponseT> ChunkUploadResponse<ResponseT> create(
       boolean isComplete, @Nullable ResponseT response) {
+    return create(isComplete, response, null);
+  }
+
+  public static <ResponseT> ChunkUploadResponse<ResponseT> create(
+      boolean isComplete, @Nullable ResponseT response, @Nullable String uploadStatus) {
     return new AutoValue_ChunkUploadResponse.Builder<ResponseT>()
         .setComplete(isComplete)
         .setResponse(response)
+        .setUploadStatus(uploadStatus)
         .build();
   }
 
@@ -75,6 +87,8 @@
 
     public abstract Builder<ResponseT> setResponse(@Nullable ResponseT response);
 
+    public abstract Builder<ResponseT> setUploadStatus(@Nullable String uploadStatus);
+
     public abstract ChunkUploadResponse<ResponseT> build();
   }
 }