This is an automated email from the ASF dual-hosted git repository.
mmarshall pushed a commit to branch branch-2.10
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.10 by this push:
new 52ca27cdde1 [fix][client] Fix potentially unfinished CompletableFuture
in doReconsumeLater (#14947)
52ca27cdde1 is described below
commit 52ca27cdde1701e2e65a71763fae3e68226ab133
Author: Qiang Zhao <[email protected]>
AuthorDate: Wed Apr 6 02:26:06 2022 +0800
[fix][client] Fix potentially unfinished CompletableFuture in
doReconsumeLater (#14947)
### Motivation
As the code is shown below, if the future returned by `doAcknowledge` is
completed exceptionally, the ``result`` future can't complete.
```java
typedMessageBuilderNew.sendAsync()
.thenAccept(__ -> doAcknowledge(finalMessageId, ackType,
Collections.emptyMap(), null)
.thenAccept(v -> result.complete(null)))
.exceptionally(ex -> {
result.completeExceptionally(ex);
return null;
});
```
### Modifications
- Use ``thenCompose`` to instead of ``thenAccept``.
### Verifying this change
- [x] Make sure that the change passes the CI checks.
### Documentation
- [x] `no-need-doc`
(cherry picked from commit 64f020cf699ecaebfec359b58d4b13491cb5f232)
---
.../src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java | 4 ++--
1 file changed, 2 insertions(+), 2 deletions(-)
diff --git
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
index a8487569167..ff43ab29065 100644
---
a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
+++
b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java
@@ -668,8 +668,8 @@ public class ConsumerImpl<T> extends ConsumerBase<T>
implements ConnectionHandle
typedMessageBuilderNew.key(message.getKey());
}
typedMessageBuilderNew.sendAsync()
- .thenAccept(__ -> doAcknowledge(finalMessageId,
ackType, Collections.emptyMap(), null)
- .thenAccept(v -> result.complete(null)))
+ .thenCompose(__ -> doAcknowledge(finalMessageId,
ackType, Collections.emptyMap(), null))
+ .thenAccept(v -> result.complete(null))
.exceptionally(ex -> {
result.completeExceptionally(ex);
return null;
