diff --git a/NEXT_CHANGELOG.md b/NEXT_CHANGELOG.md index 0691ed444..ebb30fd47 100644 --- a/NEXT_CHANGELOG.md +++ b/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 diff --git a/databricks-sdk-java/src/main/java/com/databricks/sdk/core/ApiClient.java b/databricks-sdk-java/src/main/java/com/databricks/sdk/core/ApiClient.java index 3cbd0d61a..b45644af6 100644 --- a/databricks-sdk-java/src/main/java/com/databricks/sdk/core/ApiClient.java +++ b/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); diff --git a/databricks-sdk-java/src/test/java/com/databricks/sdk/core/ApiClientTest.java b/databricks-sdk-java/src/test/java/com/databricks/sdk/core/ApiClientTest.java index fa1bb6a6c..c339c4823 100644 --- a/databricks-sdk-java/src/test/java/com/databricks/sdk/core/ApiClientTest.java +++ b/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();