Netty async streaming response may miss terminal signal with AsyncResponseTransformer.toPublisher()

# Bug Report: Netty `toPublisher()` Can Suppress Channel-Inactive Error Before Body Subscriber Terminal Signal

Suggested issue title:

```text
Netty async streaming response may miss terminal signal with AsyncResponseTransformer.toPublisher()
```

## Describe the bug

AWS SDK Java v2 `2.44.4` appears to contain a logic-level missing-terminal-signal risk for async streaming responses using `NettyNioAsyncHttpClient`, for example:

```java
CompletableFuture> responseFuture =
s3AsyncClient.getObject(request, AsyncResponseTransformer.toPublisher());
```

`AsyncResponseTransformer.toPublisher()` completes the client-returned future when the response object is available and the response body begins streaming. After that, the SDK user subscribes to the returned `ResponsePublisher` and waits for the body subscriber to receive either `Subscriber.onComplete()` or `Subscriber.onError(Throwable)`.

The suspected issue is that the Netty response path can mark the HTTP stream as complete before the SDK has finalized the caller-visible response. If the channel becomes inactive in that gap, the current `ResponseHandler.channelInactive()` logic can suppress the fallback error because `STREAMING_COMPLETE_KEY=true`, even though `RESPONSE_COMPLETE_KEY` has not been set to true.

The relevant state is:

```text
STREAMING_COMPLETE_KEY == true
RESPONSE_COMPLETE_KEY != true
```

In that state, Netty has observed `LastHttpContent`, but the SDK has not necessarily delivered the body subscriber terminal signal or finalized the response.

This is a potential SDK-user-visible hang: a user that has already subscribed to the returned `ResponsePublisher` may wait indefinitely for `onComplete()` or `onError(Throwable)`.

## Regression Issue

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

I do not know whether this is a regression.

## Expected Behavior

When using `AsyncResponseTransformer.toPublisher()`, once the user subscribes to the returned `ResponsePublisher`, the body subscriber should eventually receive one terminal signal:

```text
Subscriber.onComplete()
```

or:

```text
Subscriber.onError(Throwable)
```

If the channel becomes inactive before the SDK has finalized the caller-visible response, the SDK should surface an error instead of silently suppressing it.

More specifically, `STREAMING_COMPLETE_KEY=true` should not be treated as sufficient proof that the caller-visible response has completed. The safer completion condition for suppressing a channel-inactive fallback error is `RESPONSE_COMPLETE_KEY=true`.

## Current Behavior

The checked upstream code was:

```text
Repository: aws/aws-sdk-java-v2
Release: 2.44.4
Published: 2026-05-07
Commit: 2619c3d643ba0444a9ea9d9234cf119fa4968931
```

The Netty client uses two different channel attributes:

```text
STREAMING_COMPLETE_KEY = Netty observed LastHttpContent
RESPONSE_COMPLETE_KEY = SDK response was finalized
```

`HttpStreamsHandler.handleReadHttpContent` sets `STREAMING_COMPLETE_KEY=true` as soon as `LastHttpContent` is observed:

```java
boolean lastHttpContent = content instanceof LastHttpContent;
if (lastHttpContent) {
ctx.channel().attr(STREAMING_COMPLETE_KEY).set(true);
}
```

File:

```text
http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/nrs/HttpStreamsHandler.java
```

The user-visible body completion happens later through `ResponseHandler.PublisherAdapter.onComplete()`:

```java
runAndLogError(channelContext.channel(),
() -> String.format("Subscriber %s threw an exception in onComplete.", subscriber),
subscriber::onComplete);
```

and only after that does the SDK finalize the response:

```java
finalizeResponse(requestContext, channelContext);
```

`finalizeResponse` sets `RESPONSE_COMPLETE_KEY=true` and completes the HTTP execute future:

```java
channelContext.channel().attr(RESPONSE_COMPLETE_KEY).set(true);
executeFuture(channelContext).complete(null);
```

File:

```text
http-clients/netty-nio-client/src/main/java/software/amazon/awssdk/http/nio/netty/internal/ResponseHandler.java
```

However, `ResponseHandler.notifyIfResponseNotCompleted()` suppresses the channel-inactive fallback error if either `RESPONSE_COMPLETE_KEY` or `STREAMING_COMPLETE_KEY` is true:

```java
Boolean responseCompleted = handlerCtx.channel().attr(RESPONSE_COMPLETE_KEY).get();
Boolean isStreamingComplete = handlerCtx.channel().attr(STREAMING_COMPLETE_KEY).get();

if (!Boolean.TRUE.equals(responseCompleted) && !Boolean.TRUE.equals(isStreamingComplete)) {
IOException err = new IOException(NettyUtils.closedChannelMessage(handlerCtx.channel()));
runAndLogError(handlerCtx.channel(), () -> "Fail to execute SdkAsyncHttpResponseHandler#onError",
() -> requestCtx.handler().onError(err));
executeFuture(handlerCtx).completeExceptionally(err);
runAndLogError(handlerCtx.channel(), () -> "Could not release channel", () -> closeAndRelease(handlerCtx));
}
```

This means the following state suppresses the fallback error:

```text
RESPONSE_COMPLETE_KEY != true
STREAMING_COMPLETE_KEY == true
```

That condition is weaker than caller-visible completion. It only proves that Netty saw the final HTTP content; it does not prove that the returned `ResponsePublisher` delivered `onComplete()` or `onError(Throwable)` to the SDK user.

## Reproduction Steps

I do not have a deterministic standalone network reproduction. This report is based on the SDK source state machine, where the problematic state is possible by inspection.

Minimal SDK-user usage pattern:

```java
S3AsyncClient s3 = S3AsyncClient.builder()
.httpClientBuilder(NettyNioAsyncHttpClient.builder())
.build();

CompletableFuture> responseFuture =
s3.getObject(getObjectRequest, AsyncResponseTransformer.toPublisher());

ResponsePublisher responsePublisher = responseFuture.join();

CompletableFuture bodyDone = new CompletableFuture<>();
responsePublisher.subscribe(new Subscriber() {
private Subscription subscription;

@Override
public void onSubscribe(Subscription subscription) {
this.subscription = subscription;
subscription.request(Long.MAX_VALUE);
}

@Override
public void onNext(ByteBuffer byteBuffer) {
// Consume body bytes.
}

@Override
public void onError(Throwable throwable) {
bodyDone.completeExceptionally(throwable);
}

@Override
public void onComplete() {
bodyDone.complete(null);
}
});

bodyDone.join();
```

Logic-level reproduction path:

```text
1. Use an async streaming response with NettyNioAsyncHttpClient and AsyncResponseTransformer.toPublisher().
2. The SDK completes the getObject future when the response body begins streaming.
3. The user subscribes to the returned ResponsePublisher.
4. Netty receives LastHttpContent.
5. HttpStreamsHandler sets STREAMING_COMPLETE_KEY=true.
6. Before ResponseHandler.PublisherAdapter.onComplete() reaches user subscriber.onComplete()
and before finalizeResponse() sets RESPONSE_COMPLETE_KEY=true, the channel becomes inactive.
7. ResponseHandler.channelInactive() calls notifyIfResponseNotCompleted().
8. notifyIfResponseNotCompleted sees STREAMING_COMPLETE_KEY=true and RESPONSE_COMPLETE_KEY=false.
9. The fallback IOException is suppressed, so the user-visible body subscriber may receive no terminal signal.
```

The normal path should complete correctly:

```text
LastHttpContent
-> STREAMING_COMPLETE_KEY=true
-> HandlerPublisher.complete()
-> PublisherAdapter.onComplete()
-> user subscriber.onComplete()
-> finalizeResponse()
-> RESPONSE_COMPLETE_KEY=true
```

The concern is the gap between `STREAMING_COMPLETE_KEY=true` and `RESPONSE_COMPLETE_KEY=true`.

## Possible Solution

`ResponseHandler.notifyIfResponseNotCompleted()` should not treat `STREAMING_COMPLETE_KEY=true` as sufficient evidence that the caller-visible response has completed.

Safer behavior would be to surface the channel-inactive error whenever:

```text
RESPONSE_COMPLETE_KEY != true
```

At minimum, the case below should not be silent:

```text
STREAMING_COMPLETE_KEY == true
RESPONSE_COMPLETE_KEY != true
```

A regression test could explicitly set or drive this state and call `channelInactive()`, then assert that the SDK surfaces an error to the response handler, the execute future, or the user-visible body subscriber instead of suppressing the error.

## Additional Information/Context

This report does not claim that `LastHttpContent` normally prevents `subscriber.onComplete()`. In the normal path, the SDK should call `subscriber.onComplete()` and then `finalizeResponse()`.

The issue is that the current channel-inactive guard uses a lower-level Netty stream marker as if it were equivalent to caller-visible completion. These are not equivalent:

```text
STREAMING_COMPLETE_KEY=true
means Netty observed LastHttpContent.

RESPONSE_COMPLETE_KEY=true
means ResponseHandler.finalizeResponse() ran.
```

For `toPublisher()`, caller-visible body completion is the body subscriber receiving `onComplete()` or `onError(Throwable)`, not the initial `CompletableFuture>` completing.

The `ResponsePublisher` timeout does not fix this case. It only cancels the underlying publisher when the SDK user does not subscribe promptly:

```java
subscribed = true;
if (timeoutTask != null) {
timeoutTask.cancel(false);
}

publisher.subscribe(subscriber);
```

Once the user has subscribed, this timeout is cancelled and does not protect against a missing post-subscribe terminal signal.

The API call and API call attempt timers also do not close this gap by default. The async pipeline attaches those timers to the SDK operation future, and `AsyncResponseTransformer.toPublisher()` completes that future when the body publisher begins streaming. Once the `getObject(..., toPublisher())` future is complete, those operation-level timers are cancelled even though the body subscriber may continue waiting for its terminal signal.

I also did not find a targeted regression test for:

```text
STREAMING_COMPLETE_KEY == true
RESPONSE_COMPLETE_KEY != true
channelInactive()
```

## AWS Java SDK version used

```text
2.44.4
```

## JDK version used

```text
Not environment-specific. Found by source inspection of aws/aws-sdk-java-v2.
```

## Operating System and version

```text
Not environment-specific. Found by source inspection of aws/aws-sdk-java-v2.
```