You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by go...@apache.org on 2021/12/17 12:22:49 UTC
[incubator-inlong] branch master updated: [INLONG-2026][Bug] Found Pulsar client create failure when starting Sort (#2027)
This is an automated email from the ASF dual-hosted git repository.
gosonzhang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-inlong.git
The following commit(s) were added to refs/heads/master by this push:
new dd9e8f7 [INLONG-2026][Bug] Found Pulsar client create failure when starting Sort (#2027)
dd9e8f7 is described below
commit dd9e8f76f5bbc993d33197ac6748ffc03f89f4b1
Author: gosonzhang <46...@qq.com>
AuthorDate: Fri Dec 17 20:22:41 2021 +0800
[INLONG-2026][Bug] Found Pulsar client create failure when starting Sort (#2027)
---
inlong-sort/pom.xml | 2 +-
.../java/org/apache/inlong/sort/flink/pulsar/PulsarUtils.java | 9 +++++----
2 files changed, 6 insertions(+), 5 deletions(-)
diff --git a/inlong-sort/pom.xml b/inlong-sort/pom.xml
index a3894ae..2a969af 100644
--- a/inlong-sort/pom.xml
+++ b/inlong-sort/pom.xml
@@ -53,7 +53,7 @@
<snappy.version>1.1.4</snappy.version>
<hadoop.version>2.8.5</hadoop.version>
<hive.version>2.3.8</hive.version>
- <pulsar.version>2.6.2</pulsar.version>
+ <pulsar.version>2.8.1</pulsar.version>
</properties>
<modules>
diff --git a/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/flink/pulsar/PulsarUtils.java b/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/flink/pulsar/PulsarUtils.java
index a7e453a..31c346d 100644
--- a/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/flink/pulsar/PulsarUtils.java
+++ b/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/flink/pulsar/PulsarUtils.java
@@ -39,6 +39,7 @@ import java.util.concurrent.TimeUnit;
import org.apache.flink.configuration.Configuration;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.admin.PulsarAdminException;
+import org.apache.pulsar.client.admin.internal.PulsarAdminImpl;
import org.apache.pulsar.client.api.MessageId;
import org.apache.pulsar.client.api.MessageRoutingMode;
import org.apache.pulsar.client.api.Producer;
@@ -142,7 +143,7 @@ public class PulsarUtils {
public static PulsarAdmin createAdmin(
String adminUrl
) throws PulsarClientException {
- return new PulsarAdmin(adminUrl, new ClientConfigurationData());
+ return new PulsarAdminImpl(adminUrl, new ClientConfigurationData());
}
public static Set<String> getTopicPartitions(
@@ -170,10 +171,10 @@ public class PulsarUtils {
String consumerGroup
) throws PulsarAdminException {
TopicStats topicStats = admin.topics().getStats(topicPartition);
- if (topicStats.subscriptions.containsKey(consumerGroup)) {
+ if (topicStats.getSubscriptions().containsKey(consumerGroup)) {
SubscriptionStats subStats =
- topicStats.subscriptions.get(consumerGroup);
- if (subStats.consumers.size() != 0) {
+ topicStats.getSubscriptions().get(consumerGroup);
+ if (subStats.getConsumers().size() != 0) {
throw new RuntimeException("Subscription been actively used by other consumers in this situation, the "
+ "exactly-once semantics cannot be guaranteed.");
} else {