You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by rd...@apache.org on 2019/05/09 01:37:02 UTC
[pulsar] branch master updated: [pulsar-storm] provide
auto-unsubscribe option in pulsar-spout (#4238)
This is an automated email from the ASF dual-hosted git repository.
rdhabalia 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 05c99a6 [pulsar-storm] provide auto-unsubscribe option in pulsar-spout (#4238)
05c99a6 is described below
commit 05c99a688f5931800ff445eeda7e80be3b3a8c8c
Author: Rajan Dhabalia <rd...@apache.org>
AuthorDate: Wed May 8 18:36:56 2019 -0700
[pulsar-storm] provide auto-unsubscribe option in pulsar-spout (#4238)
---
.../src/main/java/org/apache/pulsar/storm/PulsarSpout.java | 5 +++++
.../org/apache/pulsar/storm/PulsarSpoutConfiguration.java | 14 ++++++++++++++
2 files changed, 19 insertions(+)
diff --git a/pulsar-storm/src/main/java/org/apache/pulsar/storm/PulsarSpout.java b/pulsar-storm/src/main/java/org/apache/pulsar/storm/PulsarSpout.java
index 2588741..5156b50 100644
--- a/pulsar-storm/src/main/java/org/apache/pulsar/storm/PulsarSpout.java
+++ b/pulsar-storm/src/main/java/org/apache/pulsar/storm/PulsarSpout.java
@@ -103,6 +103,11 @@ public class PulsarSpout extends BaseRichSpout implements IMetric {
public void close() {
try {
LOG.info("[{}] Closing Pulsar consumer for topic {}", spoutId, pulsarSpoutConf.getTopic());
+
+ if (pulsarSpoutConf.isAutoUnsubscribe()) {
+ consumer.unsubscribe();
+ }
+
if (!pulsarSpoutConf.isSharedConsumerEnabled() && consumer != null) {
consumer.close();
}
diff --git a/pulsar-storm/src/main/java/org/apache/pulsar/storm/PulsarSpoutConfiguration.java b/pulsar-storm/src/main/java/org/apache/pulsar/storm/PulsarSpoutConfiguration.java
index 8bd139b..bb8d8c8 100644
--- a/pulsar-storm/src/main/java/org/apache/pulsar/storm/PulsarSpoutConfiguration.java
+++ b/pulsar-storm/src/main/java/org/apache/pulsar/storm/PulsarSpoutConfiguration.java
@@ -45,6 +45,7 @@ public class PulsarSpoutConfiguration extends PulsarStormConfiguration {
private boolean sharedConsumerEnabled = false;
private SubscriptionType subscriptionType = SubscriptionType.Shared;
+ private boolean autoUnsubscribe = false;
/**
* @return the subscription name for the consumer in the spout
@@ -146,4 +147,17 @@ public class PulsarSpoutConfiguration extends PulsarStormConfiguration {
public void setSharedConsumerEnabled(boolean sharedConsumerEnabled) {
this.sharedConsumerEnabled = sharedConsumerEnabled;
}
+
+ public boolean isAutoUnsubscribe() {
+ return autoUnsubscribe;
+ }
+
+ /**
+ * It unsubscribes the subscription when spout gets closed in the topology.
+ *
+ * @param autoUnsubscribe
+ */
+ public void setAutoUnsubscribe(boolean autoUnsubscribe) {
+ this.autoUnsubscribe = autoUnsubscribe;
+ }
}