Permalink
Browse files

Change JUnit test to clojure.test test.

Pulled out the retry policy into its own class.
  • Loading branch information...
1 parent 330b8bd commit 61827e72fe7e715e72fba0b23e7ec86ebe66344d @d2r d2r committed Feb 20, 2013
@@ -286,36 +286,48 @@ public static long secureRandomLong() {
public static CuratorFramework newCurator(Map conf, List<String> servers, Object port, String root) {
return newCurator(conf, servers, port, root, null);
}
-
+
+ public static class BoundedExponentialBackoffRetry extends ExponentialBackoffRetry {
+
+ protected final int maxRetryInterval;
+
+ public BoundedExponentialBackoffRetry(int baseSleepTimeMs,
+ int maxRetries, int maxSleepTimeMs) {
+ super(baseSleepTimeMs, maxRetries);
+ this.maxRetryInterval = maxSleepTimeMs;
+ }
+
+ public int getMaxRetryInterval() {
+ return this.maxRetryInterval;
+ }
+
+ @Override
+ public int getSleepTimeMs(int count, long elapsedMs)
+ {
+ return Math.min(maxRetryInterval,
+ super.getSleepTimeMs(count, elapsedMs));
+ }
+
+ }
+
public static CuratorFramework newCurator(Map conf, List<String> servers, Object port, String root, ZookeeperAuthInfo auth) {
List<String> serverPorts = new ArrayList<String>();
for(String zkServer: (List<String>) servers) {
serverPorts.add(zkServer + ":" + Utils.getInt(port));
}
- String zkStr = StringUtils.join(serverPorts, ",") + root;
+ 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(
+ .retryPolicy(new BoundedExponentialBackoffRetry(
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));
- }
- });
+ Utils.getInt(conf.get(Config.STORM_ZOOKEEPER_RETRY_TIMES)),
+ Utils.getInt(conf.get(Config.STORM_ZOOKEEPER_RETRY_INTERVAL_CEILING))));
if(auth!=null && auth.scheme!=null) {
builder = builder.authorization(auth.scheme, auth.payload);
- }
+ }
return builder.build();
} catch (IOException e) {
throw new RuntimeException(e);
@@ -0,0 +1,33 @@
+(ns backtype.storm.util-test
+ (:import [backtype.storm Config])
+ (:import [backtype.storm.utils Utils])
+ (:import [com.netflix.curator.retry ExponentialBackoffRetry])
+ (:use [backtype.storm util])
+ (:use [clojure test])
+)
+
+(deftest test-new-curator-uses-exponential-backoff
+ (let [expected_interval 2400
+ expected_retries 10
+ expected_ceiling 5000
+ conf (merge (clojurify-structure (Utils/readDefaultConfig))
+ {Config/STORM_ZOOKEEPER_RETRY_INTERVAL expected_interval
+ Config/STORM_ZOOKEEPER_RETRY_TIMES expected_retries
+ Config/STORM_ZOOKEEPER_RETRY_INTERVAL_CEILING expected_ceiling})
+ servers ["bogus_server"]
+ arbitrary_port 42
+ curator (Utils/newCurator conf servers arbitrary_port)
+ retry (-> curator .getZookeeperClient .getRetryPolicy)
+ ]
+ (is (.isAssignableFrom ExponentialBackoffRetry (.getClass retry)))
+ (is (= (.getBaseSleepTimeMs retry) expected_interval))
+ (is (= (.getN retry) expected_retries))
+ (is (= (.getMaxRetryInterval retry) expected_ceiling))
+ ; It would be very unlikely for this to fail three times.
+ (is (or
+ (= (.getSleepTimeMs retry 10 0) expected_ceiling)
+ (= (.getSleepTimeMs retry 10 0) expected_ceiling)
+ (= (.getSleepTimeMs retry 10 0) expected_ceiling)
+ ))
+ )
+)
@@ -1,45 +0,0 @@
-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);
- }
-}

0 comments on commit 61827e7

Please sign in to comment.