commit 3dc4ddf2575299aad2c93abce02e8ca0f5d9256b
author Richard Yu <yohan.richard.yu@gmail.com> Tue Apr 02 09:41:04 2019 -0700
committer Sijie Guo <guosijie@gmail.com> Tue Apr 02 09:41:04 2019 -0700
tree d8c60d303393347b0317f1fe424717681ef2c167
parent 394ae522738169b163878ee6fc2d5b9f7d0349eb

[Issue-2122] [pulsar-client] Adding configuration for backoff strategy (#3848)


Fixes #2122 

### Motivation

Current backoff strategy is set by default and is too aggressive. What we should do is allow it to be configurable by the user.

### Documentation

  - Does this pull request introduce a new feature? (yes)
  - If yes, how is the feature documented? (not sure)

pulsar-storm/src/main/java/org/apache/pulsar/storm/PulsarSpout.java

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 713007b..8907fc3 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
@@ -179,8 +179,9 @@
         if (msg != null) {
             MessageRetries messageRetries = pendingMessageRetries.get(msg.getMessageId());
             if (Backoff.shouldBackoff(messageRetries.getTimeStamp(), TimeUnit.NANOSECONDS,
-                    messageRetries.getNumRetries())) {
-                Utils.sleep(100);
+                    messageRetries.getNumRetries(), clientConf.getDefaultBackoffIntervalNanos(), 
+                    clientConf.getMaxBackoffIntervalNanos())) {
+                Utils.sleep(TimeUnit.NANOSECONDS.toMillis(clientConf.getDefaultBackoffIntervalNanos()));
             } else {
                 // remove the message from the queue and emit to the topology, only if it should not be backedoff
                 LOG.info("[{}] Retrying failed message {}", spoutId, msg.getMessageId());