You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2024/02/05 12:33:09 UTC
(pulsar) branch branch-2.10 updated: [fix][client] Fix ConsumerBuilderImpl#subscribe silent stuck when using pulsar-client:3.0.x with jackson-annotations prior to 2.12.0 (#21985)
This is an automated email from the ASF dual-hosted git repository.
technoboy 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 6c7e7bf609c [fix][client] Fix ConsumerBuilderImpl#subscribe silent stuck when using pulsar-client:3.0.x with jackson-annotations prior to 2.12.0 (#21985)
6c7e7bf609c is described below
commit 6c7e7bf609ce9f8ef417a30d91e0c13a7f1011dc
Author: 萧易客 <km...@live.com>
AuthorDate: Tue Jan 30 19:20:29 2024 +0800
[fix][client] Fix ConsumerBuilderImpl#subscribe silent stuck when using pulsar-client:3.0.x with jackson-annotations prior to 2.12.0 (#21985)
### Motivation
In summary, `jackson-annotations:2.12.0` or later is now required for `pulsar-client 3.0.x`, and this also applies to versions `3.1.x` and `3.2.x`.
Otherwise, `ConsumerBuilderImpl#subscribe` may become stuck without displaying any error message.
### Modifications
Modify the `whenComplete` to a combination of `thenAccept` and `exceptionally`. The modification is harmless.
---
.../apache/pulsar/client/impl/MultiTopicsConsumerImpl.java | 14 +++++++-------
1 file changed, 7 insertions(+), 7 deletions(-)
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
index 0eea33e05d8..163c2c0da11 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
@@ -995,13 +995,13 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
private void subscribeTopicPartitions(CompletableFuture<Void> subscribeResult, String topicName, int numPartitions,
boolean createIfDoesNotExist) {
- client.preProcessSchemaBeforeSubscribe(client, schema, topicName).whenComplete((schema, cause) -> {
- if (null == cause) {
- doSubscribeTopicPartitions(schema, subscribeResult, topicName, numPartitions, createIfDoesNotExist);
- } else {
- subscribeResult.completeExceptionally(cause);
- }
- });
+ client.preProcessSchemaBeforeSubscribe(client, schema, topicName)
+ .thenAccept(schema -> {
+ doSubscribeTopicPartitions(schema, subscribeResult, topicName, numPartitions, createIfDoesNotExist);
+ }).exceptionally(cause -> {
+ subscribeResult.completeExceptionally(cause);
+ return null;
+ });
}
private void doSubscribeTopicPartitions(Schema<T> schema,