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:45:12 UTC
[pulsar] branch branch-2.9 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 branch-2.9
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.9 by this push:
new 4a92407226d [fix][client] Fix potentially unfinished CompletableFuture in doReconsumeLater (#14947)
4a92407226d is described below
commit 4a92407226dc7923dd9c1f5110a2fe38c45540a8
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)
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;
});
```
- Use ``thenCompose`` to instead of ``thenAccept``.
- [x] Make sure that the change passes the CI checks.
- [x] `no-need-doc`
(cherry picked from commit 64f020cf699ecaebfec359b58d4b13491cb5f232)
---
.../src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
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 d19b980a24a..ff2612beee4 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
@@ -624,7 +624,8 @@ public class ConsumerImpl<T> extends ConsumerBase<T> implements ConnectionHandle
typedMessageBuilderNew.key(message.getKey());
}
typedMessageBuilderNew.sendAsync()
- .thenAccept(__ -> doAcknowledge(finalMessageId, ackType, properties, null).thenAccept(v -> result.complete(null)))
+ .thenCompose(__ -> doAcknowledge(finalMessageId, ackType, properties, null))
+ .thenAccept(v -> result.complete(null))
.exceptionally(ex -> {
result.completeExceptionally(ex);
return null;