fix: improve 503 handling for json resumable uploads (#2289)

Author: BenWhitehead

google-cloud-storage/src/main/java/com/google/cloud/storage/JsonResumableSessionPutTask.java

@@ -190,14 +190,17 @@ public void rewindTo(long offset) {
       } else {
         HttpResponseException cause = new HttpResponseException(response);
         String contentType = response.getHeaders().getContentType();
+        Long contentLength = response.getHeaders().getContentLength();
         // If the content-range header value has run ahead of the backend, it will respond with
         // a 503 with plain text content
         // Attempt to detect this very loosely as to minimize impact of modified error message
         // This is accurate circa 2023-06
         if ((!JsonResumableSessionFailureScenario.isOk(code)
                 && !JsonResumableSessionFailureScenario.isContinue(code))
             && contentType != null
-            && contentType.startsWith("text/plain")) {
+            && contentType.startsWith("text/plain")
+            && contentLength != null
+            && contentLength > 0) {
           String errorMessage = cause.getContent().toLowerCase(Locale.US);
           if (errorMessage.contains("content-range")) {
             StorageException se =

google-cloud-storage/src/test/java/com/google/cloud/storage/ITJsonResumableSessionPutTaskTest.java

@@ -740,6 +740,37 @@ public void scenario5() throws Exception {
     }
   }
 
+  @Test
+  public void _503_emptyBody() throws Exception {
+    HttpRequestHandler handler =
+        req -> {
+          FullHttpResponse resp =
+              new DefaultFullHttpResponse(req.protocolVersion(), APPEND_GREATER_THAN_CURRENT_SIZE);
+          resp.headers().set(CONTENT_TYPE, "text/plain; charset=utf-8");
+          return resp;
+        };
+
+    try (FakeHttpServer fakeHttpServer = FakeHttpServer.of(handler);
+        TmpFile tmpFile =
+            DataGenerator.base64Characters().tempFile(temp.newFolder().toPath(), _256KiBL)) {
+      URI endpoint = fakeHttpServer.getEndpoint();
+      String uploadUrl = String.format("%s/upload/%s", endpoint.toString(), UUID.randomUUID());
+
+      AtomicLong confirmedBytes = new AtomicLong(-1L);
+
+      JsonResumableSessionPutTask task =
+          new JsonResumableSessionPutTask(
+              httpClientContext,
+              uploadUrl,
+              RewindableContent.of(tmpFile.getPath()),
+              HttpContentRange.of(ByteRangeSpec.explicit(_512KiBL, _768KiBL)));
+
+      StorageException se = assertThrows(StorageException.class, task::call);
+      assertThat(se.getCode()).isEqualTo(503);
+      assertThat(confirmedBytes.get()).isEqualTo(-1);
+    }
+  }
+
   @Test
   public void jsonParseFailure() throws Exception {