Use a ceiling on ExponentialBackoffRetry

Trying to avoid pulling in a newer netflix curator Adds lein-junit and junit dependencies with test. Adds new configurable maximum ceiling for the retry interval, defaults to 30s.
nathanmarz · Feb 17, 2013 · 330b8bd · 330b8bd
1 parent e34feda
commit 330b8bd
Show file tree

Hide file tree

Showing 5 changed files with 74 additions and 4 deletions.
diff --git a/conf/defaults.yaml b/conf/defaults.yaml
@@ -14,6 +14,7 @@ storm.zookeeper.session.timeout: 20000
 storm.zookeeper.connection.timeout: 15000
 storm.zookeeper.retry.times: 5
 storm.zookeeper.retry.interval: 1000
+storm.zookeeper.retry.intervalceiling: 30000
 storm.cluster.mode: "distributed" # can be distributed or local
 storm.local.mode.zmq: false
 

diff --git a/project.clj b/project.clj
@@ -29,16 +29,22 @@
                  ]
 
   :source-paths ["src/clj"]
-  :java-source-paths ["src/jvm"]
+  :java-source-paths ["src/jvm" "test/jvm"]
   :test-paths ["test/clj"]
   :resource-paths ["conf"]
 
-  :profiles {:dev {:resource-paths ["src/dev"]}
+  :profiles {:dev {:resource-paths ["src/dev"]
+                   :dependencies [[junit/junit "4.11"]]
+                  }
              :release {}
              :lib {}
              }
 
-  :plugins [[lein-swank "1.4.4"]]
+  :plugins [[lein-swank "1.4.4"]
+            [lein-junit "1.1.2"]
+           ]
+
+  :junit ["test/jvm"]
 
   :repositories {"sonatype"
                  "http://oss.sonatype.org/content/groups/public/"}

diff --git a/src/jvm/backtype/storm/Config.java b/src/jvm/backtype/storm/Config.java
@@ -106,6 +106,11 @@ public class Config extends HashMap<String, Object> {
      */
     public static String STORM_ZOOKEEPER_RETRY_INTERVAL="storm.zookeeper.retry.interval";
 
+    /**
+     * The ceiling of the interval between retries of a Zookeeper operation.
+     */
+    public static String STORM_ZOOKEEPER_RETRY_INTERVAL_CEILING="storm.zookeeper.retry.intervalceiling";
+
     /**
      * The Zookeeper authentication scheme to use, e.g. "digest". Defaults to no authentication.
      */

diff --git a/src/jvm/backtype/storm/utils/Utils.java b/src/jvm/backtype/storm/utils/Utils.java
@@ -295,11 +295,24 @@ public static CuratorFramework newCurator(Map conf, List<String> servers, Object
         String zkStr = StringUtils.join(serverPorts, ",") + root; 
         try {
 
+            final int maxRetryInterval = Utils.getInt( 
+                    conf.get(Config.STORM_ZOOKEEPER_RETRY_INTERVAL_CEILING));
+
             CuratorFrameworkFactory.Builder builder = CuratorFrameworkFactory.builder()
                     .connectString(zkStr)
                     .connectionTimeoutMs(Utils.getInt(conf.get(Config.STORM_ZOOKEEPER_CONNECTION_TIMEOUT)))
                     .sessionTimeoutMs(Utils.getInt(conf.get(Config.STORM_ZOOKEEPER_SESSION_TIMEOUT)))
-                    .retryPolicy(new ExponentialBackoffRetry(Utils.getInt(conf.get(Config.STORM_ZOOKEEPER_RETRY_TIMES)), Utils.getInt(conf.get(Config.STORM_ZOOKEEPER_RETRY_INTERVAL))));
+                    .retryPolicy(new ExponentialBackoffRetry(
+                                Utils.getInt(conf.get(Config.STORM_ZOOKEEPER_RETRY_INTERVAL)),
+                                Utils.getInt(conf.get(Config.STORM_ZOOKEEPER_RETRY_TIMES)))
+                                {
+                                    @Override
+                                    protected int getSleepTimeMs(int count, long elapsedMs)
+                                    {
+                                        return Math.min(maxRetryInterval,
+                                                super.getSleepTimeMs(count, elapsedMs));
+                                    }
+                                });
             if(auth!=null && auth.scheme!=null) {
                 builder = builder.authorization(auth.scheme, auth.payload);
             }            

diff --git a/test/jvm/backtype/storm/utils/UtilsTest.java b/test/jvm/backtype/storm/utils/UtilsTest.java
@@ -0,0 +1,45 @@
+package backtype.storm.utils;
+
+import org.junit.Test;
+
+import java.util.List;
+import java.util.ArrayList;
+import java.util.Map;
+import java.util.HashMap;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import com.netflix.curator.CuratorZookeeperClient;
+import com.netflix.curator.framework.CuratorFramework;
+import com.netflix.curator.retry.ExponentialBackoffRetry;
+
+import backtype.storm.Config;
+
+public class UtilsTest {
+
+  @Test
+  public void testNewCuratorUsesBoundedExponentialBackoff() {
+    @SuppressWarnings("unchecked")
+    Map<String,Object> conf = (Map<String,Object>)Utils.readDefaultConfig();
+
+    // Ensure these two values are different.
+    final int ArbitraryInterval = 24;
+    final int ArbitraryRetries = 4;
+    conf.put(Config.STORM_ZOOKEEPER_RETRY_INTERVAL, ArbitraryInterval);
+    conf.put(Config.STORM_ZOOKEEPER_RETRY_TIMES, ArbitraryRetries);
+
+    List<String> servers = new ArrayList<String>();
+    servers.add("bogus_server");
+    Object port = new Integer(42);
+    CuratorFramework cf = Utils.newCurator(conf, servers, port);
+
+    assertTrue(cf.getZookeeperClient().getRetryPolicy() 
+            instanceof ExponentialBackoffRetry);
+
+    ExponentialBackoffRetry retry =
+        (ExponentialBackoffRetry)cf.getZookeeperClient().getRetryPolicy();
+    assertEquals(retry.getBaseSleepTimeMs(), ArbitraryInterval);
+    assertEquals(retry.getN(), ArbitraryRetries);
+  }
+}