Do not retry requests with a consumed streaming body (#872)

## Summary

Requests with a streaming body (e.g. `files().upload()`) are backed by a
single-use
`InputStream`. When such a request received a retriable HTTP response
(429/501/503), the
retry loop in `ApiClient.executeInner` re-sent the *same* `InputStream`.
Because the first
attempt had already consumed the stream, the retry transmitted an
**empty body**, which
manifested as either:

- a **0-byte file** written to the target despite an HTTP 204, or
- a confusing downstream error (e.g. `InternalError`) surfaced from
`FilesImpl.upload`.

This only occurs when a retriable status interrupts a streaming upload,
so it is rare and
intermittent, but when it happens the upload silently corrupts data.

## Root cause

`Request` holds the streaming body as a single `InputStream` reference.
On retry,
`executeInner` reuses the same `Request`, and `CommonsHttpClient` wraps
that already-consumed
stream in a fresh `InputStreamEntity` — sending 0 bytes. The upload
`PUT` is non-idempotent,
and `NonIdempotentRequestRetryStrategy` treats 429/501/503 as retriable,
so a transient 503
triggers the empty-body retry.

An `InputStream`-backed body cannot be rewound, so retrying is
inherently unsafe once the body
has been sent.

## Fix

In `ApiClient.executeInner`, skip the retry when the request has a
streaming body **and** an
HTTP response was received (which proves the body was already
transmitted), and surface the
original error to the caller so it can retry with a fresh stream.

The guard is deliberately narrow:

- **String / no-body requests** (the vast majority of API calls) are
repeatable — a new entity
  is built per attempt — so their retry behavior is unchanged.
- **Transport-level `IOError`s** (no response received, e.g. a pre-send
`ConnectException`)
still retry as before, since in that case the stream may not have been
read yet.

The only behavior removed is the retry of an already-consumed stream,
which never worked. This
lives in the hand-written core (`ApiClient`), so it applies to any
streaming upload rather than
being specific to the Files API.

## Tests

- `doesNotRetryStreamingBodyAfterResponse` — a streaming `PUT` that gets
a 503 now throws the
typed `TemporarilyUnavailable` (503) instead of silently retrying with
an empty body.
- `retriesNonStreamingBodyOn503` — a string-bodied request with the same
503 still retries and
  succeeds, confirming the guard does not regress ordinary requests.

Full module test suite passes (1404 tests, 0 failures).

Author: Divyansh-db

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,20 @@ 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) means
+      // the body was already sent, so retrying would re-send an empty stream and upload 0 bytes.
+      // The stream cannot be rewound, so surface the original error and let the caller retry with a
+      // fresh stream. Transport IOErrors (response == null) still retry, as the body may be unsent.
+      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,7 +6,9 @@
 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.HttpClient;
 import com.databricks.sdk.core.http.Request;
 import com.databricks.sdk.core.http.Response;
 import com.databricks.sdk.core.utils.FakeTimer;
@@ -15,10 +17,13 @@
 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.io.InputStream;
 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 +489,107 @@ void privateLinkRedirectBecomesPrivateLinkValidationError() throws MalformedURLE
     assertTrue(e.getMessage().contains("AWS PrivateLink"));
   }
 
+  /**
+   * A fake HttpClient that reads the request body to EOF on every call, mirroring how the real
+   * CommonsHttpClient drains the entity onto the wire. It records the number of body bytes actually
+   * transmitted per attempt, so tests can assert what a retry would (or would not) send. The status
+   * code returned for each attempt is supplied up front.
+   */
+  private static class BodyReadingHttpClient implements HttpClient {
+    private final Deque<Integer> statusCodes;
+    final List<Integer> bytesReadPerAttempt = new ArrayList<>();
+
+    BodyReadingHttpClient(Integer... statusCodesInOrder) {
+      this.statusCodes = new ArrayDeque<>(Arrays.asList(statusCodesInOrder));
+    }
+
+    @Override
+    public Response execute(Request in) throws IOException {
+      // The SDK issues a best-effort GET /.well-known/databricks-config host-metadata pre-flight
+      // through this same client before the request under test. Ignore it: return a benign 404
+      // (the SDK falls back to user config) without recording it or consuming a status code.
+      if (in.getUrl().contains("/.well-known/")) {
+        return new Response(in, 404, "Not Found", Collections.emptyMap());
+      }
+      int total = 0;
+      if (in.isBodyStreaming() && in.getBodyStream() != null) {
+        InputStream is = in.getBodyStream();
+        byte[] buf = new byte[4096];
+        int r;
+        while ((r = is.read(buf)) != -1) {
+          total += r;
+        }
+      } else if (in.isBodyString() && in.getBodyString() != null) {
+        total = in.getBodyString().getBytes(StandardCharsets.UTF_8).length;
+      }
+      bytesReadPerAttempt.add(total);
+      int status = statusCodes.isEmpty() ? 204 : statusCodes.removeFirst();
+      String reason = EnglishReasonPhraseCatalog.INSTANCE.getReason(status, Locale.ENGLISH);
+      return new Response(in, status, reason, Collections.emptyMap());
+    }
+  }
+
+  private ApiClient apiClientWith(HttpClient httpClient) {
+    DatabricksConfig config =
+        new DatabricksConfig()
+            .setHost("http://my.host")
+            .setCredentialsProvider(new DummyCredentialsProvider())
+            .setHttpClient(httpClient);
+    return new ApiClient(config, new FakeTimer());
+  }
+
+  @Test
+  void doesNotRetryStreamingBodyAfterResponse() throws IOException {
+    // A streaming body (e.g. Files.upload) is a single-use InputStream consumed by the first
+    // attempt, so retrying a 503 would re-send an empty stream and silently upload 0 bytes. The
+    // upload must be attempted exactly once and the 503 surfaced to the caller.
+    byte[] contents = "file-contents".getBytes(StandardCharsets.UTF_8);
+    // The 204 is what a buggy empty-body retry would receive; the guard means it is never reached.
+    BodyReadingHttpClient hc = new BodyReadingHttpClient(503, 204);
+    ApiClient client = apiClientWith(hc);
+
+    InputStream body = new ByteArrayInputStream(contents);
+    // Catch rather than assertThrows so the byte-count assertions below run first: without the fix
+    // no exception is thrown, and those assertions give the more informative [13, 0] failure.
+    DatabricksError thrown = null;
+    try {
+      client.execute(new Request("PUT", "/api/2.0/fs/files/Volumes/c/s/v/f", body), Void.class);
+    } catch (DatabricksError e) {
+      thrown = e;
+    }
+
+    // Exactly one attempt, sending the full body. Without the guard this is [13, 0]: a full first
+    // attempt followed by an empty-body retry.
+    assertEquals(
+        1,
+        hc.bytesReadPerAttempt.size(),
+        "streaming upload must not be retried; bytes sent per attempt: " + hc.bytesReadPerAttempt);
+    assertEquals(contents.length, hc.bytesReadPerAttempt.get(0));
+    assertEquals(-1, body.read(), "stream is single-use and should be fully consumed");
+    assertInstanceOf(TemporarilyUnavailable.class, thrown);
+    assertEquals(503, thrown.getStatusCode());
+  }
+
+  @Test
+  void retriesNonStreamingBodyOn503AndResendsFullBody() throws IOException {
+    // Complement to doesNotRetryStreamingBodyAfterResponse: a string-bodied request is repeatable,
+    // so the same 503 must still be retried, and crucially the retry must re-send the full body
+    // (a fresh entity is built per attempt). This confirms the streaming guard is scoped narrowly
+    // and does not regress ordinary requests.
+    String jsonBody = "{\"key\":\"value\"}";
+    BodyReadingHttpClient hc = new BodyReadingHttpClient(503, 200);
+    ApiClient client = apiClientWith(hc);
+
+    client.execute(
+        new Request("POST", "/api/2.0/sql/statements/", jsonBody), MyEndpointResponse.class);
+
+    // The 503 was retried, and both attempts sent the full body (a string body is re-sendable).
+    int expected = jsonBody.getBytes(StandardCharsets.UTF_8).length;
+    assertEquals(2, hc.bytesReadPerAttempt.size());
+    assertEquals(expected, hc.bytesReadPerAttempt.get(0));
+    assertEquals(expected, hc.bytesReadPerAttempt.get(1));
+  }
+
   @Test
   void testDefaultWorkspaceIdReturnsNullWhenNotSet() {
     Request req = getBasicRequest();