Fix NPE in AsyncRetryableStage (#6936)

When retrying and checking for the 'x-amz-retry-after' or 'retry-after'
header on the response, ensure that the response is not null first.

Author: dagnir

.changes/next-release/bugfix-AWSSDKforJavav2-857b9f2.json

@@ -0,0 +1,6 @@
+{
+    "type": "bugfix",
+    "category": "AWS SDK for Java v2",
+    "contributor": "",
+    "description": "Fix an issue in the async SDK clients where a retry can lead to a `NullPointerException` if the exception that the SDK encountered did not originate from the service, such as a connection exception."
+}

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AsyncRetryableStage.java

@@ -192,6 +192,10 @@ private Duration suggestedDelay() {
      * Returns the suggested backoff delay based on the 'x-amz-retry-after' header value in the response.
      */
     private Optional<Duration> xAmzRetryAfter(SdkHttpResponse response) {
+        if (response == null) {
+            return Optional.empty();
+        }
+
         Optional<String> optionalXAmzRetryAfter = response.firstMatchingHeader(X_AMZ_RETRY_AFTER_HEADER);
         return optionalXAmzRetryAfter.map(xAmzRetryAfter -> {
             try {

core/sdk-core/src/test/java/software/amazon/awssdk/core/internal/http/pipeline/stages/AsyncRetryableStageTest.java

@@ -22,6 +22,7 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+import java.io.IOException;
 import java.net.URI;
 import java.time.Duration;
 import java.util.concurrent.CompletableFuture;
@@ -33,6 +34,7 @@
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
 import org.junit.jupiter.params.provider.MethodSource;
 import org.mockito.ArgumentCaptor;
 import software.amazon.awssdk.core.Response;
@@ -233,4 +235,67 @@ void execute_retryableException_treatsRetryAfterCorrectly(RetryAfterTestCase tes
 
         assertThat(refreshRequest.suggestedDelay().get()).isEqualTo(testCase.expectedDelay());
     }
+
+    @ParameterizedTest(name = "New Retries = {0}")
+    @CsvSource({"true", "false"})
+    void execute_delegateThrows_noHttpResponse_uses0SuggestedDelay(boolean newRetries2026) throws Exception {
+        SdkClientConfiguration clientConfig = SdkClientConfiguration.builder()
+                                                                    .option(SdkClientOption.RETRY_STRATEGY, mockRetryStrategy)
+                                                                    .option(SdkClientOption.SCHEDULED_EXECUTOR_SERVICE,
+                                                                            executorService)
+                                                                    .build();
+
+        HttpClientDependencies deps = HttpClientDependencies.builder()
+                                                            .clientConfiguration(clientConfig)
+                                                            .build();
+
+        AsyncRetryableStage<SdkResponse> retryableStage = new AsyncRetryableStage<>(mock(TransformingAsyncResponseHandler.class),
+                                                                                    deps, mockDelegatePipeline);
+
+        SdkHttpFullRequest httpRequest = SdkHttpFullRequest.builder()
+                                                           .method(SdkHttpMethod.GET)
+                                                           .uri(URI.create("https://my-service.amazonaws.com"))
+                                                           .build();
+
+        ExecutionAttributes execAttrs = ExecutionAttributes.builder()
+                                                           .put(SdkInternalExecutionAttribute.NEW_RETRIES_2026_ENABLED,
+                                                                newRetries2026)
+                                                           .build();
+
+        ExecutionContext execCtx = ExecutionContext.builder()
+                                                   .metricCollector(NoOpMetricCollector.create())
+                                                   .executionAttributes(execAttrs)
+                                                   .build();
+
+        RequestExecutionContext ctx = RequestExecutionContext.builder()
+                                                             .originalRequest(mock(SdkRequest.class))
+                                                             .executionContext(execCtx)
+                                                             .build();
+
+        SdkHttpFullResponse.Builder httpResponse = SdkHttpFullResponse.builder()
+                                                                      .statusCode(502);
+
+        Response<SdkResponse> response = Response.<SdkResponse>builder()
+                                                 .httpResponse(httpResponse.build())
+                                                 .isSuccess(false)
+                                                 .exception(SdkException.builder().build())
+                                                 .build();
+
+
+        CompletableFuture<Response<SdkResponse>> future = new CompletableFuture<>();
+        future.completeExceptionally(new IOException("connection"));
+        when(mockDelegatePipeline.execute(any(), any())).thenReturn(future);
+
+        CompletableFuture<Response<SdkResponse>> execute = retryableStage.execute(httpRequest, ctx);
+        // exception thrown doesn't matter, just results in exception because we mock just enough...
+        assertThatThrownBy(execute::join);
+
+        ArgumentCaptor<RefreshRetryTokenRequest> refreshRequestCaptor = ArgumentCaptor.forClass(RefreshRetryTokenRequest.class);
+
+        verify(mockRetryStrategy).refreshRetryToken(refreshRequestCaptor.capture());
+
+        RefreshRetryTokenRequest refreshRequest = refreshRequestCaptor.getValue();
+
+        assertThat(refreshRequest.suggestedDelay().get()).isEqualTo(Duration.ZERO);
+    }
 }