[pulsar] branch master updated: [pulsar-storm] allow option to configure queue size of pulsar-spout-consumer (#4239)

[mm...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

mmerli 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 815c8fa  [pulsar-storm] allow option to configure queue size of pulsar-spout-consumer (#4239)
815c8fa is described below

commit 815c8fa2facbdf17797e78828428e12ce35babb4
Author: Rajan Dhabalia <rd...@apache.org>
AuthorDate: Tue May 14 23:44:39 2019 -0700

    [pulsar-storm] allow option to configure queue size of pulsar-spout-consumer (#4239)
---
 .../src/main/java/org/apache/pulsar/storm/PulsarSpout.java |  1 +
 .../org/apache/pulsar/storm/PulsarSpoutConfiguration.java  | 14 ++++++++++++++
 2 files changed, 15 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 5156b50..bbfe5cd 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
@@ -93,6 +93,7 @@ public class PulsarSpout extends BaseRichSpout implements IMetric {
         this.consumerConf.setTopicNames(Collections.singleton(pulsarSpoutConf.getTopic()));
         this.consumerConf.setSubscriptionName(pulsarSpoutConf.getSubscriptionName());
         this.consumerConf.setSubscriptionType(pulsarSpoutConf.getSubscriptionType());
+        this.consumerConf.setReceiverQueueSize(pulsarSpoutConf.getConsumerReceiverQueueSize());
 
         this.pulsarSpoutConf = pulsarSpoutConf;
         this.failedRetriesTimeoutNano = pulsarSpoutConf.getFailedRetriesTimeout(TimeUnit.NANOSECONDS);
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 bb8d8c8..0f27ddc 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
@@ -46,6 +46,7 @@ public class PulsarSpoutConfiguration extends PulsarStormConfiguration {
 
     private SubscriptionType subscriptionType = SubscriptionType.Shared;
     private boolean autoUnsubscribe = false;
+    private int consumerReceiverQueueSize = 1000;
 
     /**
      * @return the subscription name for the consumer in the spout
@@ -71,6 +72,19 @@ public class PulsarSpoutConfiguration extends PulsarStormConfiguration {
         this.subscriptionType = subscriptionType;
     }
 
+    public int getConsumerReceiverQueueSize() {
+        return consumerReceiverQueueSize;
+    }
+
+    /**
+     * Receiver queue size of pulsar-spout consumer.
+     * 
+     * @param consumerReceiverQueueSize
+     */
+    public void setConsumerReceiverQueueSize(int consumerReceiverQueueSize) {
+        this.consumerReceiverQueueSize = consumerReceiverQueueSize;
+    }
+    
     /**
      * @return the mapper to convert pulsar message to a storm tuple
      */