Skip to content

Handle exceptions thrown by the consumer passed to SdkPublisher.subscribe or filter. #2877

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Dec 1, 2021
Merged

Handle exceptions thrown by the consumer passed to SdkPublisher.subscribe or filter. #2877

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2ce1a6e
Handle exceptions thrown by the consumer passed to SdkPublisher.subsc…
millems Nov 30, 2021
cd3ec38
Merge branch 'master' into millem/paginator-hang
millems Dec 1, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions .changes/next-release/bugfix-AWSSDKforJavav2-ef61843.json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,6 @@
{
"category": "AWS SDK for Java v2",
"contributor": "",
"type": "bugfix",
"description": "Complete the future returned by SdkPublisher.subscribe or filter exceptionally if the subscriber or predicate throws an exception."
}
59 changes: 59 additions & 0 deletions core/sdk-core/src/test/java/software/amazon/awssdk/core/async/SdkPublishersTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -16,12 +16,17 @@
package software.amazon.awssdk.core.async;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import org.junit.Test;
import org.reactivestreams.Publisher;
import org.reactivestreams.Subscriber;
Expand Down Expand Up @@ -84,6 +89,60 @@ public void mapHandlesError() {
assertThat(fakeSubscriber.recordedErrors()).containsExactly(exception);
}

@Test
public void subscribeHandlesError() {
FakeSdkPublisher<String> fakePublisher = new FakeSdkPublisher<>();
RuntimeException exception = new IllegalArgumentException("Failure!");

CompletableFuture<Void> subscribeFuture = fakePublisher.subscribe(s -> {
throw exception;
});

fakePublisher.publish("one");
fakePublisher.complete();

assertThat(subscribeFuture.isCompletedExceptionally()).isTrue();
assertThatThrownBy(() -> subscribeFuture.get(5, TimeUnit.SECONDS))
.isInstanceOf(ExecutionException.class)
.hasCause(exception);
}

@Test
public void filterHandlesError() {
FakeSdkPublisher<String> fakePublisher = new FakeSdkPublisher<>();
RuntimeException exception = new IllegalArgumentException("Failure!");

CompletableFuture<Void> subscribeFuture = fakePublisher.filter(s -> {
throw exception;
}).subscribe(r -> {});

fakePublisher.publish("one");
fakePublisher.complete();

assertThat(subscribeFuture.isCompletedExceptionally()).isTrue();
assertThatThrownBy(() -> subscribeFuture.get(5, TimeUnit.SECONDS))
.isInstanceOf(ExecutionException.class)
.hasCause(exception);
}

@Test
public void flatMapIterableHandlesError() {
FakeSdkPublisher<String> fakePublisher = new FakeSdkPublisher<>();
RuntimeException exception = new IllegalArgumentException("Failure!");

CompletableFuture<Void> subscribeFuture = fakePublisher.flatMapIterable(s -> {
throw exception;
}).subscribe(r -> {});

fakePublisher.publish("one");
fakePublisher.complete();

assertThat(subscribeFuture.isCompletedExceptionally()).isTrue();
assertThatThrownBy(() -> subscribeFuture.get(5, TimeUnit.SECONDS))
.isInstanceOf(ExecutionException.class)
.hasCause(exception);
}

private final static class FakeByteBufferSubscriber implements Subscriber<ByteBuffer> {
private final List<String> recordedEvents = new ArrayList<>();

Expand Down
46 changes: 46 additions & 0 deletions ...namodb/src/it/java/software/amazon/awssdk/services/dynamodb/PaginatorIntegrationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,10 +24,15 @@
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.BiFunction;
import java.util.stream.Stream;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.Test;
import software.amazon.awssdk.core.async.SdkPublisher;
import software.amazon.awssdk.core.pagination.sync.SdkIterable;
import software.amazon.awssdk.services.dynamodb.model.AttributeDefinition;
import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
Expand All @@ -43,6 +48,7 @@
import software.amazon.awssdk.services.dynamodb.model.ScanRequest;
import software.amazon.awssdk.services.dynamodb.model.ScanResponse;
import software.amazon.awssdk.services.dynamodb.paginators.ScanIterable;
import software.amazon.awssdk.services.dynamodb.paginators.ScanPublisher;
import utils.resources.tables.BasicTempTable;
import utils.test.util.DynamoDBTestBase;
import utils.test.util.TableUtils;
Expand Down Expand Up @@ -211,6 +217,46 @@ public void mix_Iterator_And_Stream_Calls() {
}
assertEquals(ITEM_COUNT, count);
}

@Test
public void sdkPublisher_subscribe_handlesExceptions() throws Exception {
RuntimeException innerException = new RuntimeException();
ScanRequest request = scanRequest(2);
try {
dynamoAsync.scanPaginator(request).subscribe(r -> {
throw innerException;
}).get(5, TimeUnit.SECONDS);
} catch (ExecutionException executionException) {
assertThat(executionException.getCause()).isSameAs(innerException);
}
}

@Test
public void sdkPublisher_filter_handlesExceptions() {
sdkPublisherFunctionHandlesException((p, t) -> p.filter(f -> { throw t; }));
}

@Test
public void sdkPublisher_map_handlesExceptions() {
sdkPublisherFunctionHandlesException((p, t) -> p.map(f -> { throw t; }));
}

@Test
public void sdkPublisher_flatMapIterable_handlesExceptions() {
sdkPublisherFunctionHandlesException((p, t) -> p.flatMapIterable(f -> { throw t; }));
}

public void sdkPublisherFunctionHandlesException(BiFunction<ScanPublisher, RuntimeException, SdkPublisher<?>> function) {
RuntimeException innerException = new RuntimeException();
ScanRequest request = scanRequest(2);
try {
function.apply(dynamoAsync.scanPaginator(request), innerException).subscribe(r -> {}).get(5, TimeUnit.SECONDS);
} catch (ExecutionException executionException) {
assertThat(executionException.getCause()).isSameAs(innerException);
} catch (InterruptedException | TimeoutException e) {
throw new AssertionError("SDK Publisher function did not handle exceptions correctly.", e);
}
}

private static void putTestData() {
Map<String, AttributeValue> item = new HashMap();
Expand Down
5 changes: 4 additions & 1 deletion services/dynamodb/src/test/java/utils/test/util/DynamoDBTestBase.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,10 +24,10 @@
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

import software.amazon.awssdk.awscore.exception.AwsServiceException;
import software.amazon.awssdk.core.exception.SdkClientException;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.dynamodb.DynamoDbAsyncClient;
import software.amazon.awssdk.services.dynamodb.DynamoDbClient;
import software.amazon.awssdk.services.dynamodb.model.AttributeValue;
import software.amazon.awssdk.services.dynamodb.model.DescribeTableRequest;
Expand All @@ -43,6 +43,8 @@ public class DynamoDBTestBase extends AwsTestBase {

protected static DynamoDbClient dynamo;

protected static DynamoDbAsyncClient dynamoAsync;

private static final Logger log = Logger.loggerFor(DynamoDBTestBase.class);

public static void setUpTestBase() {
Expand All @@ -53,6 +55,7 @@ public static void setUpTestBase() {
}

dynamo = DynamoDbClient.builder().region(REGION).credentialsProvider(CREDENTIALS_PROVIDER_CHAIN).build();
dynamoAsync = DynamoDbAsyncClient.builder().region(REGION).credentialsProvider(CREDENTIALS_PROVIDER_CHAIN).build();
}

public static DynamoDbClient getClient() {
Expand Down
16 changes: 11 additions & 5 deletions utils/src/main/java/software/amazon/awssdk/utils/async/FilteringSubscriber.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -40,11 +40,17 @@ public void onSubscribe(Subscription subscription) {

@Override
public void onNext(T t) {
if (predicate.test(t)) {
subscriber.onNext(t);
} else {
// Consumed a demand but didn't deliver. Request other to make up for it
subscription.request(1);
try {
if (predicate.test(t)) {
subscriber.onNext(t);
} else {
// Consumed a demand but didn't deliver. Request other to make up for it
subscription.request(1);
}
} catch (RuntimeException e) {
// Handle the predicate throwing an exception
subscription.cancel();
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Query just for my curiosity : Earlier these exceptions were thrown higher in stack and now we are catching it , do we need to worry about any backward compatibility issue were Users have expecting this exceptions to be caught directly.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's against "the rules" to throw out of an onNext (except for null parameters), so we're out of compliance already here. It's possible someone is relying on the old behavior: https://xkcd.com/1172/

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks Matt.

onError(e);
}
}
}
10 changes: 8 additions & 2 deletions utils/src/main/java/software/amazon/awssdk/utils/async/SequentialSubscriber.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,8 +46,14 @@ public void onSubscribe(Subscription subscription) {

@Override
public void onNext(T t) {
consumer.accept(t);
subscription.request(1);
try {
consumer.accept(t);
subscription.request(1);
} catch (RuntimeException e) {
// Handle the consumer throwing an exception
subscription.cancel();
future.completeExceptionally(e);
}
}

@Override
Expand Down