You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by mm...@apache.org on 2022/04/05 18:26:14 UTC
[pulsar] branch master updated: [fix][client] Fix potentially unfinished CompletableFuture in doReconsumeLater (#14947)
This is an automated email from the ASF dual-hosted git repository.
mmarshall pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new 64f020cf699 [fix][client] Fix potentially unfinished CompletableFuture in doReconsumeLater (#14947)
64f020cf699 is described below
commit 64f020cf699ecaebfec359b58d4b13491cb5f232
Author: Qiang Zhao <74...@users.noreply.github.com>
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`
---
.../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 5c5f48f7d1a..74bda5e4e91 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
@@ -667,8 +667,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;