public ConfigOption(String nameSpace,String zkUrls){
    this(
            nameSpace,
            zkUrls,
            new ExponentialBackoffRetry(60000, 3),
            DefaultOptions.USE_REMOTE_CONFIG);
}
 

@Before
public void setUp() throws Exception {
    server = new TestingServer();
    String connectionString = server.getConnectString();
    Map conf = new HashMap();
    conf.put(Config.STORM_ZOOKEEPER_SESSION_TIMEOUT, 1000);
    conf.put(Config.STORM_ZOOKEEPER_RETRY_TIMES, 4);
    conf.put(Config.STORM_ZOOKEEPER_RETRY_INTERVAL, 5);
    ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3);
    zookeeper = CuratorFrameworkFactory.newClient(connectionString, retryPolicy);
    dynamicBrokersReader = new DynamicBrokersReader(conf, connectionString, masterPath, topic);
    zookeeper.start();
}
 

private void initZk() throws IOException {
	this.zkClient = createWithOptions(zkAddress,
			new ExponentialBackoffRetry(1000, 3), 1000, 1000);
	this.zkClient.start();
}
 

@Bean( initMethod = "start", destroyMethod = "close" )
public CuratorFramework curator() {
	 return CuratorFrameworkFactory.newClient( ZK_HOST, new ExponentialBackoffRetry( 1000, 3 ) );
}
 

@Bean( initMethod = "start", destroyMethod = "close" )
public CuratorFramework curator() {
	 return CuratorFrameworkFactory.newClient( ZK_HOST, new ExponentialBackoffRetry( 1000, 3 ) );
}