The subscription of pagination silently stops when `Subscriber.onNext` occurred an error
- Dominant language
- Java
- Stars
- 2.6k
- Forks
- 1k
- Avg merge
- 2d 9h
- Merged PRs (30d)
- 51
Description
### Describe the bug
I encounted the `reactive-streams` for pagination did not complete by the `DynamoDbAsyncClient.queryPaginator`.
I found that if an exception occurred in the `onNext` method of the `Subscriber` passed to the Publisher returned by the `DynamoDbAsyncClient.queryPaginator`, then the `CompletableFuture` has failed and did not work in the following point.
- https://github.com/aws/aws-sdk-java-v2/blob/2.19.4/core/sdk-core/src/main/java/software/amazon/awssdk/core/pagination/async/ResponsesSubscription.java#L61
- https://github.com/aws/aws-sdk-java-v2/blob/2.19.4/core/sdk-core/src/main/java/software/amazon/awssdk/core/internal/pagination/async/ItemsSubscription.java#L86
### Expected Behavior
The `FlatteningSubscriber` and `MappingSubscriber` that this SDK have are logging, cancel subscription, pass an exception to `Subscriber.onError` and throw it.
- https://github.com/aws/aws-sdk-java-v2/blob/2.19.4/utils/src/main/java/software/amazon/awssdk/utils/async/FlatteningSubscriber.java#L110-L113
- https://github.com/aws/aws-sdk-java-v2/blob/2.19.4/utils/src/main/java/software/amazon/awssdk/utils/internal/MappingSubscriber.java#L71-L76
I think it should behave similarly to these for consistency of behavior within the SDK.
**_Additional informations_**
The Reactive manifest has the following:
https://github.com/reactive-streams/reactive-streams-jvm/blob/v1.0.4/README.md#specification
(related issue: https://github.com/reactive-streams/reactive-streams-jvm/issues/297)
> 2.13
> Calling `onSubscribe`, `onNext`, `onError` or `onComplete` MUST return normally except when any provided parameter is `null` in which case it MUST throw a `java.lang.NullPointerException` to the caller, for all other situations the only legal way for a `Subscriber` to signal failure is by cancelling its `Subscription`. In the case that this rule is violated, any associated `Subscription` to the `Subscriber` MUST be considered as cancelled, and the caller MUST raise this error condition in a fashion that is adequate for the runtime environment.
> *The intent of this rule is to establish the semantics for the methods of Subscriber and what the Publisher is allowed to do in which case this rule is violated. «Raise this error condition in a fashion that is adequate for the runtime environment» could mean logging the error—or otherwise make someone or something aware of the situation—as the error cannot be signalled to the faulty Subscriber.*
The `java.util.current.Flow.Subscriber` has the following:
https://docs.oracle.com/javase/9/docs/api/java/util/concurrent/Flow.Subscriber.html
> void onNext(T item)
> Method invoked with a Subscription's next item. If this method throws an exception, resulting behavior is not guaranteed, but may cause the Subscription to be cancelled.
### Current Behavior
It is not logging, does not cancel subscription and does not other behave.
### Reproduction Steps
I reproduces the bug in the following repository.
https://github.com/hayasshi/aws-sdk-java-v2-issues-poc
If you want to run it, please read the README.md.
https://github.com/hayasshi/aws-sdk-java-v2-issues-poc/blob/main/README.md
There is the code that reproduce the issue.
https://github.com/hayasshi/aws-sdk-java-v2-issues-poc/blob/main/src/test/scala/example/AwsSdkIssuesPoc.scala#L93
### Possible Solution
_No response_
### Additional Information/Context
_No response_
### AWS Java SDK version used
2.19.4
### JDK version used
openjdk version "1.8.0_352" OpenJDK Runtime Environment (Temurin)(build 1.8.0_352-b08) OpenJDK 64-Bit Server VM (Temurin)(build 25.352-b08, mixed mode)
### Operating System and version
macOS 12.6.2
Contributor guide
Research direction
Start with core/sdk-core/src/main/java/software/amazon/awssdk/core/pagination/async/ResponsesSubscription.java and core/internal/pagination/async/ItemsSubscription.java, then compare their behavior with FlatteningSubscriber.java and MappingSubscriber.java. Run the reproduction described in the linked README and inspect AwsSdkIssuesPoc.scala at the indicated line. Done means an exception from Subscriber.onNext is logged, the subscription is cancelled, and the failure is handled consistently with the other subscribers.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- backend
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Stale
- Clarity
- Mostly clear
- Newbie friendliness
- 45/100