This is an automated email from the ASF dual-hosted git repository.
mmarshall pushed a commit to branch branch-2.8
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.8 by this push:
new 9925a9d856b [fix][client] Fix potentially unfinished CompletableFuture
in doReconsumeLater (#14947)
9925a9d856b is described below
commit 9925a9d856bd4d34641310064f6b801e0c315f00
Author: Qiang Zhao <[email protected]>
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 ebf3583c227..8104123e4c3 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
@@ -616,7 +616,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;
