Fixed an issue where async waiters failed to match wrapped exception

Author: zoewangg

.changes/next-release/bugfix-AWSSDKforJavav2-fa98630.json

@@ -0,0 +1,6 @@
+{
+    "category": "AWS SDK for Java v2", 
+    "contributor": "", 
+    "type": "bugfix", 
+    "description": "Fixed an issue where an async waiter failed to match exception when the exception was wrapped with CompletionException. See [#2460](https://github.com/aws/aws-sdk-java-v2/issues/2460)"
+}

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/waiters/AsyncWaiterExecutor.java

@@ -18,6 +18,7 @@
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
@@ -71,7 +72,17 @@ private void runAsyncPollingFunction(Supplier<CompletableFuture<T>> asyncPolling
                                          long startTime) {
         asyncPollingFunction.get().whenComplete((response, exception) -> {
             try {
-                Either<T, Throwable> responseOrException = exception == null ? Either.left(response) : Either.right(exception);
+                Either<T, Throwable> responseOrException;
+
+                if (exception == null) {
+                    responseOrException = Either.left(response);
+                } else {
+                    if (exception instanceof CompletionException) {
+                        responseOrException = Either.right(exception.getCause());
+                    } else {
+                        responseOrException = Either.right(exception);
+                    }
+                }
 
                 Optional<WaiterAcceptor<? super T>> optionalWaiterAcceptor =
                     executorHelper.firstWaiterAcceptorIfMatched(responseOrException);

core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/waiters/WaiterExecutorHelper.java

@@ -90,7 +90,7 @@ public Either<Long, SdkClientException> nextDelayOrUnretryableException(int atte
 
     public SdkClientException noneMatchException(Either<T, Throwable> responseOrException) {
         return responseOrException.map(
-            r -> SdkClientException.create("No acceptor was matched for the response"),
+            r -> SdkClientException.create("No acceptor was matched for the response: " + r),
             t -> SdkClientException.create("An exception was thrown and did not match any "
                                            + "waiter acceptors", t));
     }

core/sdk-core/src/test/java/software/amazon/awssdk/core/waiters/AsyncWaiterTest.java

@@ -20,6 +20,7 @@
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionException;
 import java.util.function.BiFunction;
 import java.util.function.Supplier;
 import org.junit.Test;
@@ -48,6 +49,17 @@ public BiFunction<Integer, TestWaiterConfiguration, WaiterResponse<String>> succ
                                                           .runAsync(new ThrowExceptionResource(count)).join();
     }
 
+    @Test
+    public void matchException_exceptionIsWrapped_shouldReturnUnwrappedException() {
+        TestWaiterConfiguration waiterConfig = new TestWaiterConfiguration()
+            .overrideConfiguration(p -> p.maxAttempts(3).backoffStrategy(BackoffStrategy.none()))
+            .addAcceptor(WaiterAcceptor.successOnExceptionAcceptor(s -> s.getMessage().contains(SUCCESS_STATE_MESSAGE)));
+
+        WaiterResponse<String> response = successOnWrappedExceptionWaiterOperation().apply(1, waiterConfig);
+        assertThat(response.matched().exception().get()).isExactlyInstanceOf(RuntimeException.class);
+        assertThat(response.attemptsExecuted()).isEqualTo(1);
+    }
+
     @Test
     public void missingScheduledExecutor_shouldThrowException() {
         assertThatThrownBy(() -> AsyncWaiter.builder(String.class)
@@ -91,6 +103,15 @@ public void requestOverrideConfig_shouldTakePrecedence() {
                            .join()).hasMessageContaining("exceeded the max retry attempts: 1");
     }
 
+    private BiFunction<Integer, TestWaiterConfiguration, WaiterResponse<String>> successOnWrappedExceptionWaiterOperation() {
+        return (count, waiterConfiguration) -> AsyncWaiter.builder(String.class)
+                                                          .overrideConfiguration(waiterConfiguration.getPollingStrategy())
+                                                          .acceptors(waiterConfiguration.getWaiterAcceptors())
+                                                          .scheduledExecutorService(executorService)
+                                                          .build()
+                                                          .runAsync(new ThrowCompletionExceptionResource(count)).join();
+    }
+
     private static final class ReturnResponseResource implements Supplier<CompletableFuture<String>> {
         private final int successAttemptIndex;
         private int count;
@@ -125,6 +146,23 @@ public CompletableFuture<String> get() {
 
             return CompletableFutureUtils.failedFuture(new RuntimeException(SUCCESS_STATE_MESSAGE));
         }
+    }
 
+    private static final class ThrowCompletionExceptionResource implements Supplier<CompletableFuture<String>> {
+        private final int successAttemptIndex;
+        private int count;
+
+        public ThrowCompletionExceptionResource(int successAttemptIndex) {
+            this.successAttemptIndex = successAttemptIndex;
+        }
+
+        @Override
+        public CompletableFuture<String> get() {
+            if (++count < successAttemptIndex) {
+                return CompletableFutureUtils.failedFuture(new CompletionException(new RuntimeException(NON_SUCCESS_STATE_MESSAGE)));
+            }
+
+            return CompletableFutureUtils.failedFuture(new CompletionException(new RuntimeException(SUCCESS_STATE_MESSAGE)));
+        }
     }
 }

services/dynamodb/src/it/java/software/amazon/awssdk/services/dynamodb/WaitersIntegrationTest.java

@@ -16,10 +16,8 @@
 package software.amazon.awssdk.services.dynamodb;
 
 import static org.assertj.core.api.Assertions.assertThat;
-import static org.junit.Assert.assertTrue;
 
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ExecutionException;
 import org.assertj.core.api.Condition;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
@@ -30,7 +28,6 @@
 import software.amazon.awssdk.services.dynamodb.model.DeleteTableRequest;
 import software.amazon.awssdk.services.dynamodb.model.DescribeTableRequest;
 import software.amazon.awssdk.services.dynamodb.model.DescribeTableResponse;
-import software.amazon.awssdk.services.dynamodb.model.DynamoDbException;
 import software.amazon.awssdk.services.dynamodb.model.KeySchemaElement;
 import software.amazon.awssdk.services.dynamodb.model.KeyType;
 import software.amazon.awssdk.services.dynamodb.model.ProvisionedThroughput;
@@ -70,10 +67,10 @@ public static void setUp() {
 
     @AfterClass
     public static void cleanUp() {
-        dynamo.deleteTable(DeleteTableRequest.builder().tableName(TABLE_NAME).build());
+        dynamoAsync.deleteTable(DeleteTableRequest.builder().tableName(TABLE_NAME).build());
 
         WaiterResponse<DescribeTableResponse> waiterResponse =
-            dynamo.waiter().waitUntilTableNotExists(b -> b.tableName(TABLE_NAME));
+            dynamoAsync.waiter().waitUntilTableNotExists(b -> b.tableName(TABLE_NAME)).join();
 
         assertThat(waiterResponse.matched().response()).isEmpty();
         assertThat(waiterResponse.matched().exception()).hasValueSatisfying(

services/s3/src/it/java/software/amazon/awssdk/services/s3/GetObjectAsyncIntegrationTest.java

@@ -63,7 +63,9 @@ public static void setupFixture() throws IOException {
         s3Async.putObject(PutObjectRequest.builder()
                                           .bucket(BUCKET)
                                           .key(KEY)
-                                          .build(), file.toPath()).join();
+                                          .build(), file.toPath());
+
+        s3Async.waiter().waitUntilObjectExists(b -> b.bucket(BUCKET).key(KEY)).join();
     }
 
     @AfterClass