Do not retry requests with a consumed streaming body

NEXT_CHANGELOG.md

@@ -8,6 +8,8 @@
 
 ### Bug Fixes
 
+* Fixed requests with a streaming body (e.g. `files().upload()`) silently uploading an empty body when retried. A single-use `InputStream` body is consumed by the first attempt, so retrying a retriable error (e.g. HTTP 503) re-sent an empty stream, which could write a 0-byte file or surface as a confusing error. The SDK no longer retries a streaming request once its body has been sent, and instead surfaces the original error so the caller can retry with a fresh stream.
+
 ### Security Vulnerabilities
 
 ### Documentation

databricks-sdk-java/src/main/java/com/databricks/sdk/core/ApiClient.java

@@ -275,6 +275,22 @@ private Response executeInner(Request in, String path, RequestOptions options) {
       if (!retryStrategy.isRetriable(databricksError)) {
         throw databricksError;
       }
+
+      // A streaming request body (e.g. Files.upload) is backed by a single-use InputStream that the
+      // first attempt consumes as it is sent. Receiving an HTTP response (response != null) proves
+      // the body was already transmitted, so retrying would re-send an empty body and silently
+      // upload 0 bytes (or surface as a confusing downstream error). Since the stream cannot be
+      // rewound, surface the original error instead so the caller can retry with a fresh stream.
+      // Transport-level IOErrors (response == null, e.g. a pre-send ConnectException) are left to
+      // retry as before, since in that case the stream may not have been read.
+      if (in.isBodyStreaming() && response != null) {
+        LOG.debug(
+            "Not retrying {} despite a retriable error: the request has a non-repeatable streaming"
+                + " body that was already consumed by the previous attempt",
+            in.getRequestLine());
+        throw databricksError;
+      }
+
       if (attemptNumber == maxAttempts) {
         throw new DatabricksException(
             String.format("Request %s failed after %d retries", in, maxAttempts), databricksError);

databricks-sdk-java/src/test/java/com/databricks/sdk/core/ApiClientTest.java

@@ -6,6 +6,7 @@
 import com.databricks.sdk.core.error.PrivateLinkValidationError;
 import com.databricks.sdk.core.error.details.ErrorDetails;
 import com.databricks.sdk.core.error.details.ErrorInfo;
+import com.databricks.sdk.core.error.platform.TemporarilyUnavailable;
 import com.databricks.sdk.core.error.platform.TooManyRequests;
 import com.databricks.sdk.core.http.Request;
 import com.databricks.sdk.core.http.Response;
@@ -15,10 +16,12 @@
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.errorprone.annotations.CanIgnoreReturnValue;
+import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.net.UnknownHostException;
+import java.nio.charset.StandardCharsets;
 import java.time.*;
 import java.util.*;
 import org.apache.http.impl.EnglishReasonPhraseCatalog;
@@ -484,6 +487,45 @@ void privateLinkRedirectBecomesPrivateLinkValidationError() throws MalformedURLE
     assertTrue(e.getMessage().contains("AWS PrivateLink"));
   }
 
+  @Test
+  void doesNotRetryStreamingBodyAfterResponse() throws IOException {
+    // Regression test: a streaming request body (e.g. Files.upload) is backed by a single-use
+    // InputStream that the first attempt consumes. Receiving a retriable HTTP response means the
+    // body was already sent, so a retry would upload an empty body. Verify the client surfaces the
+    // original error to the caller instead of retrying.
+    String path = "/api/2.0/fs/files/Volumes/main/default/vol/f.json";
+    String url = "http://my.host" + path;
+    byte[] contents = "file-contents".getBytes(StandardCharsets.UTF_8);
+    Request stub = new Request("PUT", url, new ByteArrayInputStream(contents));
+    // If the guard fails and a retry is issued, it would consume the success response and pass.
+    ApiClient client =
+        getApiClient(
+            stub,
+            Arrays.asList(getTransientError(stub, 503, (String) null), getSuccessResponse(stub)));
+
+    ByteArrayInputStream body = new ByteArrayInputStream(contents);
+    DatabricksError exception =
+        assertThrows(
+            DatabricksError.class,
+            () -> client.execute(new Request("PUT", path, body), Void.class));
+
+    assertInstanceOf(TemporarilyUnavailable.class, exception);
+    assertEquals(503, exception.getStatusCode());
+  }
+
+  @Test
+  void retriesNonStreamingBodyOn503() throws IOException {
+    // Complement to doesNotRetryStreamingBodyAfterResponse: a string-bodied request is repeatable
+    // (a fresh entity is built per attempt), so the same 503 must still be retried as before. This
+    // confirms the streaming guard is scoped narrowly and does not regress ordinary requests.
+    Request req = getExampleNonIdempotentRequest();
+    runApiClientTest(
+        req,
+        Arrays.asList(getTransientError(req, 503, (String) null), getSuccessResponse(req)),
+        MyEndpointResponse.class,
+        new MyEndpointResponse().setKey("value"));
+  }
+
   @Test
   void testDefaultWorkspaceIdReturnsNullWhenNotSet() {
     Request req = getBasicRequest();