S3AsyncClient.getObject() with toPublisher() does not retry read timeouts during body streaming

### Describe the bug

> **Note:** No existing GitHub issue covers this specific behavior. The closest related issues are #5158 (backpressure) and #6198 (upload retries), but neither addresses the retry gap for streaming response transformers.

When using `S3AsyncClient.getObject()` with `AsyncResponseTransformer.toPublisher()`, the SDK's built-in retry mechanism does not retry **mid-stream errors** — including retryable S3 503 responses and read timeouts that occur during body streaming.

The root cause is that `PublisherAsyncResponseTransformer.onStream()` calls `future.complete()` as soon as the response body stream becomes available (after HTTP headers arrive), which causes `AsyncRetryableStage` to consider the request successful and close the retry window. Any error that occurs afterward during body streaming (e.g., S3 503-in-body, Netty `ReadTimeoutException`) is delivered via `Publisher.onError()` but is never seen by the retry handler.

This behavior is **not documented as a retry limitation** — the Javadoc for `toPublisher()` only mentions that the future completes when streaming begins, without noting the retry consequence.

### Regression Issue

- [ ] Select this option if this issue appears to be a regression.

### Expected Behavior

Mid-stream retryable errors (503, read timeouts, connection resets) should be retried by the SDK's retry policy, consistent with the behavior of `toBytes()` and `toFile()`.

At minimum, the Javadoc for `toPublisher()` should explicitly warn that SDK retries do not cover errors that occur during body streaming.

### Current Behavior

- S3 returns a 503 error mid-stream → `S3Exception` with `SDK Attempt Count: 1` (no retry)

- Netty read timeout during body streaming → `IOException: Read timed out` delivered via `Publisher.onError()` (no retry)

- The `CompletableFuture` returned by `getObject()` completes successfully at header-time, so any retry wrapper around the future is also ineffective

## Error Evidence

### Error 1: S3 503 with no retry

```
software.amazon.awssdk.services.s3.model.S3Exception:
(Service: S3, Status Code: 503, Request ID: xxxxxxx,
Extended Request ID: xxxxxxxxxxxxxxxxxx)
(SDK Attempt Count: 1)
```

Note: `SDK Attempt Count: 1` confirms zero retries despite 503 being a retryable status code.

### Error 2: Read timeout during body streaming

```
Caused by: java.io.IOException: Read timed out
Channel Information: ChannelDiagnostics(channel=[id: 0x73df4585, ...],
channelAge=PT38.440909521S, requestCount=51, responseCount=51,
lastIdleDuration=PT0.139699762S)
at ...ResponseHandler$PublisherAdapter$1.onError(ResponseHandler.java:345)
at ...HandlerPublisher.exceptionCaught(HandlerPublisher.java:473)
...
Caused by: io.netty.handler.timeout.ReadTimeoutException
```

Note: Error is delivered via `PublisherAdapter.onError()` — outside the retry boundary.

**Important:** This also affects `toBlockingInputStream()`, which completes its future when streaming begins (not ends). Only `toBytes()` and `toFile()` keep the future pending until the full body is consumed.

## Root Cause Analysis (Source Code)

Traced through 6 SDK source files:

### 1\. `PublisherAsyncResponseTransformer.onStream()` — Future completes at header-time

```java
// core/sdk-core/src/main/java/software/amazon/awssdk/core/async/
public void onStream(SdkPublisher publisher) {
// Completes IMMEDIATELY when stream becomes available
future.complete(new ResponsePublisher<>(response, publisher));
}
```

### 2\. `AsyncRetryableStage.attemptExecute()` — Retry handler hooks into that future

```java
// core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/
responseFuture.whenComplete((response, ex) -> {
// For toPublisher(), this fires at header-time
// Retry handler sees "success" and closes retry window
});
```

### 3\. `PublisherAsyncResponseTransformer.exceptionOccurred()` — Late errors are a NO-OP

```java
public void exceptionOccurred(Throwable error) {
// future.completeExceptionally(error) — but future was already
// completed in onStream(), so this is a NO-OP
future.completeExceptionally(error);
}
```

### Flow diagram

```
toPublisher():
Request → Headers arrive → onStream() → future.complete() → RETRY WINDOW CLOSES
↓
Body streaming → Error occurs mid-stream
↓
exceptionOccurred() → future.completeExceptionally() → NO-OP
↓
Publisher.onError() → Error escapes to subscriber → No retry

toBytes() (for comparison):
Request → Headers arrive → onStream() → subscribes, future STILL PENDING
↓
Body streaming → All bytes received
→ future.complete()
OR
Body streaming → Error mid-stream
→ future.completeExceptionally()
↓
Retry handler sees failure → RETRY ✅
```

### Reproduction Steps

Reproduction Steps

1. Use S3AsyncClient with default retry policy

2. Call getObject(request, AsyncResponseTransformer.toPublisher())

3. Subscribe to the returned publisher and consume the body stream

4. Trigger a mid-stream error (e.g., large file on a loaded S3 prefix to provoke 503, or set a very short readTimeout)

5. Observe: SDK Attempt Count: 1 — no retry occurs

Minimal Reproduction (Java)

The key challenge in reproducing this is that a very short readTimeout (e.g. 1ms) fires before headers arrive, causing the CompletableFuture itself to fail — which IS correctly retried. The bug only manifests when the timeout occurs after headers arrive but during body streaming.

Strategy: Use a readTimeout long enough for headers to arrive (2s), then delay subscribing to the body publisher for longer than the timeout (4s). This guarantees:

1. Headers arrive → onStream() fires → future completes → retry window closes

2. No data is read for 4s → Netty ReadTimeoutHandler fires after 2s

3. ReadTimeoutException delivered via Publisher.onError() — no retry

We compare with toBytes() using a 1ms timeout, where the same type of error IS retried.

Prerequisites: An S3 bucket with a >1MB object, AWS credentials configured.

import software.amazon.awssdk.core.async.AsyncResponseTransformer;

import software.amazon.awssdk.core.async.SdkPublisher;

import software.amazon.awssdk.core.async.ResponsePublisher;

import software.amazon.awssdk.core.async.SdkPublisher;

import software.amazon.awssdk.core.client.config.ClientOverrideConfiguration;

import software.amazon.awssdk.core.retry.RetryPolicy;

import software.amazon.awssdk.http.nio.netty.NettyNioAsyncHttpClient;

import software.amazon.awssdk.services.s3.S3AsyncClient;

import software.amazon.awssdk.services.s3.model.GetObjectRequest;

import software.amazon.awssdk.services.s3.model.GetObjectResponse;


import java.nio.ByteBuffer;

import java.time.Duration;

import java.util.concurrent.CompletableFuture;

import java.util.concurrent.atomic.AtomicLong;

import org.reactivestreams.Subscriber;

import org.reactivestreams.Subscription;

public class S3PublisherRetryReproduction {

    private static final String BUCKET = "your-bucket";

    private static final String KEY    = "your-key";

    private static final Duration READ_TIMEOUT = Duration.ofSeconds(2);

    private static final long SUBSCRIBE_DELAY_MS = 4000; // > READ_TIMEOUT

    public static void main(String[] args) throws Exception {

        System.out.println("=== Test 1: toPublisher() — mid-stream timeout, NO retry ===\n");

        testToPublisher();

        System.out.println("\n=== Test 2: toBytes() — retries work correctly ===\n");

        testToBytes();

    }

    /** toPublisher(): future completes at header-time → no retry for body errors */

    static void testToPublisher() throws Exception {

        try (S3AsyncClient client = S3AsyncClient.builder()

                .httpClient(NettyNioAsyncHttpClient.builder()

                    .readTimeout(READ_TIMEOUT).build())

                .overrideConfiguration(ClientOverrideConfiguration.builder()

                    .retryPolicy(RetryPolicy.builder().numRetries(3).build())

                    .build())

                .build()) {

            GetObjectRequest req = GetObjectRequest.builder()

                .bucket(BUCKET).key(KEY).build();

            // Future completes when headers arrive — retry window closes

            ResponsePublisher<GetObjectResponse> pub = client.getObject(req,

                AsyncResponseTransformer.<GetObjectResponse>toPublisher()).join();

            System.out.println("✅ Future completed (headers). Retry window CLOSED.");

            System.out.println("⏳ Delaying " + SUBSCRIBE_DELAY_MS + "ms to trigger read timeout...\n");

            Thread.sleep(SUBSCRIBE_DELAY_MS);

            // Subscribe after timeout — error arrives via onError(), not retry handler

            AtomicLong bytes = new AtomicLong(0);

            CompletableFuture<Void> done = new CompletableFuture<>();

            pub.subscribe(new Subscriber<ByteBuffer>() {

                Subscription sub;

                public void onSubscribe(Subscription s) {

                    sub = s; s.request(Long.MAX_VALUE);

                }

                public void onNext(ByteBuffer b) {

                    bytes.addAndGet(b.remaining());

                }

                public void onError(Throwable t) {

                    System.out.println("❌ Publisher.onError() — OUTSIDE retry boundary:");

                    System.out.println("   " + t.getClass().getSimpleName()

                        + ": " + t.getMessage());

                    System.out.println("   ⚠️ No retry — SDK considers request successful");

                    done.completeExceptionally(t);

                }

                public void onComplete() { done.complete(null); }

            });

            done.join();

        } catch (Exception e) {

            // Expected — the point is the lack of retry

        }

    }

    /** toBytes(): Future stays pending until all data received → retries work */

    static void testToBytes() throws Exception {

        try (S3AsyncClient client = S3AsyncClient.builder()

                .httpClient(NettyNioAsyncHttpClient.builder()

                    .readTimeout(Duration.ofMillis(1)).build())  // 1ms to force failure

                .overrideConfiguration(ClientOverrideConfiguration.builder()

                    .retryPolicy(RetryPolicy.builder().numRetries(3).build())

                    .build())

                .build()) {

            var resp = client.getObject(GetObjectRequest.builder()

                .bucket(BUCKET).key(KEY).build(), AsyncResponseTransformer.toBytes()).join();

            System.out.println("✅ All bytes received: " + resp.asByteArray().length);

        } catch (Exception e) {

            System.out.println("❌ After retries exhausted: " + e.getCause().getMessage());

            System.out.println("   ✅ SDK retried (Attempt Count > 1)");

        }

    }

}

Expected output:

=== Test 1: toPublisher() — mid-stream timeout, NO retry ===

✅ Future completed (headers). Retry window CLOSED.

⏳ Delaying 4000ms to trigger read timeout...


❌ Publisher.onError() — OUTSIDE retry boundary:

   IOException: Read timed out

   ⚠️ No retry — SDK considers request successful

=== Test 2: toBytes() — retries work correctly ===

❌ After retries exhausted: <error message> (SDK Attempt Count: 4)

   ✅ SDK retried (Attempt Count > 1)

Affected Transformers

Transformer | Future completes when | Mid-stream retries?
-- | -- | --
toBytes() | All bytes received | ✅ Yes
toFile(Path) | All bytes written to disk | ✅ Yes
toPublisher() | Stream becomes available (headers) | ❌ No
toBlockingInputStream() | Stream becomes available (headers) | ❌ No

### Possible Solution

1. **New transformer variant**: `AsyncResponseTransformer.toRetryablePublisher()` that defers `future.complete()` until the publisher terminates (either completes or errors). This would keep the retry window open for the full transfer.

2. **Opt-in flag on** `toPublisher()`: e.g., `.toPublisher(PublisherOptions.builder().enableRetry(true).build())`

3. **Documentation fix** (minimum): Add an explicit warning to the `toPublisher()` Javadoc noting that SDK retries do not cover mid-stream errors, and recommend `toBytes()` or a custom retry wrapper for reliability.

## Workaround

Switch to `AsyncResponseTransformer.toBytes()` (which keeps the future pending until all bytes are received), or materialize the full publisher stream inside a custom retry boundary before considering the operation complete.

### Additional Information/Context

_No response_

### AWS Java SDK version used

2.31.3

### JDK version used

openjdk 21.0.11 2026-04-21 LTS OpenJDK Runtime Environment Corretto-21.0.11.10.1 (build 21.0.11+10-LTS) OpenJDK 64-Bit Server VM Corretto-21.0.11.10.1 (build 21.0.11+10-LTS, mixed mode, sharing)

### Operating System and version

Darwin 25.6.0 Darwin Kernel Version 25.6.0: Fri Jul 31 19:18:49 PDT 2026; root:xnu-12377.161.14~5/RELEASE_ARM64_T6000 arm64

Start with PublisherAsyncResponseTransformer.onStream() and exceptionOccurred() in core/sdk-core/src/main/java/software/amazon/awssdk/core/async/, then trace AsyncRetryableStage.attemptExecute() in the HTTP pipeline stages. Run the S3PublisherRetryReproduction example and compare toPublisher() with toBytes() and toFile(). Done means the mid-stream retry behavior is resolved or the toPublisher() limitation is explicitly documented.

Written by the indexing model from the issue text.