You are viewing a plain text version of this content. The canonical link for it is here.
Posted to s4-commits@incubator.apache.org by mm...@apache.org on 2012/03/02 16:32:00 UTC
[3/3] S4-7 Robust TCP-based communication - guaranteed message
delivery - pair-wise ordering of messages with potential duplicate messages -
tolerates network glitches - uses an asynchronous, ordered-per-partition,
memory-aware, thread pool to send messa
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-comm/src/test/java/org/apache/s4/fixtures/ZkBasedClusterManagementTestModule.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/test/java/org/apache/s4/fixtures/ZkBasedClusterManagementTestModule.java b/subprojects/s4-comm/src/test/java/org/apache/s4/fixtures/ZkBasedClusterManagementTestModule.java
index a368db8..3a6c6be 100644
--- a/subprojects/s4-comm/src/test/java/org/apache/s4/fixtures/ZkBasedClusterManagementTestModule.java
+++ b/subprojects/s4-comm/src/test/java/org/apache/s4/fixtures/ZkBasedClusterManagementTestModule.java
@@ -19,19 +19,23 @@ import org.apache.s4.comm.topology.AssignmentFromZK;
import org.apache.s4.comm.topology.Cluster;
import org.apache.s4.comm.topology.Topology;
import org.apache.s4.comm.topology.TopologyFromZK;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import com.google.inject.AbstractModule;
import com.google.inject.Binder;
import com.google.inject.name.Names;
public class ZkBasedClusterManagementTestModule extends AbstractModule {
-
+ private static final Logger logger = LoggerFactory.getLogger(ZkBasedClusterManagementTestModule.class);
protected PropertiesConfiguration config = null;
private Class<? extends Emitter> emitterClass = null;
private Class<? extends Listener> listenerClass = null;
protected ZkBasedClusterManagementTestModule() {
+ this.emitterClass = TCPEmitter.class;
+ this.listenerClass = TCPListener.class;
}
protected ZkBasedClusterManagementTestModule(Class<? extends Emitter> emitterClass,
@@ -72,16 +76,17 @@ public class ZkBasedClusterManagementTestModule extends AbstractModule {
bind(Assignment.class).to(AssignmentFromZK.class);
bind(Topology.class).to(TopologyFromZK.class);
- if (this.emitterClass != null) {
- bind(Emitter.class).to(this.emitterClass);
- } else {
- bind(Emitter.class).to(TCPEmitter.class);
- }
+ bind(Emitter.class).to(this.emitterClass);
+ bind(Listener.class).to(this.listenerClass);
- if (this.listenerClass != null) {
- bind(Listener.class).to(this.listenerClass);
- } else {
- bind(Listener.class).to(TCPListener.class);
+ bind(Integer.class).annotatedWith(Names.named("comm.retries")).toInstance(10);
+ bind(Integer.class).annotatedWith(Names.named("comm.retry_delay")).toInstance(10);
+ bind(Integer.class).annotatedWith(Names.named("comm.timeout")).toInstance(1000);
+
+ if (this.emitterClass.equals(TCPEmitter.class)) {
+ bind(Integer.class).annotatedWith(Names.named("tcp.partition.queue_size")).toInstance(256);
}
+
+ logger.info("Emitter set to - {}; Listener set to - {}", this.emitterClass, this.listenerClass);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-comm/src/test/java/org/apache/s4/fixtures/ZkBasedTest.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/test/java/org/apache/s4/fixtures/ZkBasedTest.java b/subprojects/s4-comm/src/test/java/org/apache/s4/fixtures/ZkBasedTest.java
index 02c5467..3b1e45a 100644
--- a/subprojects/s4-comm/src/test/java/org/apache/s4/fixtures/ZkBasedTest.java
+++ b/subprojects/s4-comm/src/test/java/org/apache/s4/fixtures/ZkBasedTest.java
@@ -6,6 +6,7 @@ import org.I0Itec.zkclient.IDefaultNameSpace;
import org.I0Itec.zkclient.ZkClient;
import org.I0Itec.zkclient.ZkServer;
import org.apache.s4.comm.tools.TaskSetup;
+import org.junit.After;
import org.junit.Before;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -13,15 +14,26 @@ import org.slf4j.LoggerFactory;
public abstract class ZkBasedTest {
private static final Logger logger = LoggerFactory.getLogger(ZkBasedTest.class);
private ZkServer zkServer;
+ protected String zookeeperAddress = "localhost:" + CommTestUtils.ZK_PORT;
+
+ private static final String clusterName = "s4-test-cluster";
+ protected final int numTasks;
+
+ protected ZkBasedTest() {
+ this.numTasks = 1;
+ }
+
+ protected ZkBasedTest(int numTasks) {
+ this.numTasks = numTasks;
+ }
@Before
- public void prepare() {
+ public void setupZkBasedTest() {
String dataDir = CommTestUtils.DEFAULT_TEST_OUTPUT_DIR + File.separator + "zookeeper" + File.separator + "data";
String logDir = CommTestUtils.DEFAULT_TEST_OUTPUT_DIR + File.separator + "zookeeper" + File.separator + "logs";
CommTestUtils.cleanupTmpDirs();
IDefaultNameSpace defaultNameSpace = new IDefaultNameSpace() {
-
@Override
public void createDefaultNameSpace(ZkClient zkClient) {
@@ -32,17 +44,17 @@ public abstract class ZkBasedTest {
zkServer = new ZkServer(dataDir, logDir, defaultNameSpace, CommTestUtils.ZK_PORT);
zkServer.start();
- logger.info("Starting Zookeeper Client 1");
- String zookeeperAddress = "localhost:" + CommTestUtils.ZK_PORT;
- @SuppressWarnings("unused")
- ZkClient zkClient = new ZkClient(zookeeperAddress, 10000, 10000);
-
- ZkClient zkClient2 = new ZkClient(zookeeperAddress, 10000, 10000);
- zkClient2.getCreationTime("/");
-
TaskSetup taskSetup = new TaskSetup(zookeeperAddress);
- final String clusterName = "s4-test-cluster";
taskSetup.clean(clusterName);
- taskSetup.setup(clusterName, 1);
+ taskSetup.setup(clusterName, this.numTasks);
+ }
+
+ @After
+ public void cleanupZkBasedTest() {
+ if (zkServer != null) {
+ zkServer.shutdown();
+ zkServer = null;
+ }
+ CommTestUtils.cleanupTmpDirs();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-core/src/main/java/org/apache/s4/core/Main.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/Main.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/Main.java
index ed4c0ce..39d0a49 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/core/Main.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/Main.java
@@ -1,13 +1,8 @@
package org.apache.s4.core;
import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileNotFoundException;
-import java.io.InputStream;
import java.util.Arrays;
-import org.apache.commons.configuration.ConfigurationException;
-import org.apache.commons.configuration.PropertiesConfiguration;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -29,16 +24,19 @@ public class Main {
* @param args
*/
public static void main(String[] args) {
-
- if (args.length == 0) {
- logger.info("Starting S4 node with default configuration");
- startDefaultS4Node();
- } else if (args.length == 1) {
- logger.info("Starting S4 node with custom configuration from file {}", args[0]);
- startCustomS4Node(args[0]);
- } else {
- logger.info("Starting S4 node in development mode");
- startDevelopmentMode(args);
+ try {
+ if (args.length == 0) {
+ logger.info("Starting S4 node with default configuration");
+ startDefaultS4Node();
+ } else if (args.length == 1) {
+ logger.info("Starting S4 node with custom configuration from file {}", args[0]);
+ startCustomS4Node(args[0]);
+ } else {
+ logger.info("Starting S4 node in development mode");
+ startDevelopmentMode(args);
+ }
+ } catch (Exception e) {
+ logger.error("S4 node failure", e);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-core/src/test/java/org/apache/s4/deploy/TestModule.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/test/java/org/apache/s4/deploy/TestModule.java b/subprojects/s4-core/src/test/java/org/apache/s4/deploy/TestModule.java
index f5f3912..011f203 100644
--- a/subprojects/s4-core/src/test/java/org/apache/s4/deploy/TestModule.java
+++ b/subprojects/s4-core/src/test/java/org/apache/s4/deploy/TestModule.java
@@ -57,6 +57,12 @@ public class TestModule extends AbstractModule {
bind(Topology.class).to(TopologyFromZK.class);
bind(Emitter.class).to(TCPEmitter.class);
bind(Listener.class).to(TCPListener.class);
+
+ bind(Integer.class).annotatedWith(Names.named("comm.retries")).toInstance(10);
+ bind(Integer.class).annotatedWith(Names.named("comm.retry_delay")).toInstance(10);
+ bind(Integer.class).annotatedWith(Names.named("comm.timeout")).toInstance(1000);
+
+ bind(Integer.class).annotatedWith(Names.named("tcp.partition.queue_size")).toInstance(256);
}
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-core/src/test/resources/default.s4.properties
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/test/resources/default.s4.properties b/subprojects/s4-core/src/test/resources/default.s4.properties
index 62fc7d5..d014a12 100644
--- a/subprojects/s4-core/src/test/resources/default.s4.properties
+++ b/subprojects/s4-core/src/test/resources/default.s4.properties
@@ -7,3 +7,4 @@ cluster.name = s4-test-cluster
cluster.zk_address = localhost:21810
cluster.zk_session_timeout = 10000
cluster.zk_connection_timeout = 10000
+tcp.partition.queue_size=256
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/8174f751/subprojects/s4-core/src/test/resources/org.apache.s4.deploy.s4.properties
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/test/resources/org.apache.s4.deploy.s4.properties b/subprojects/s4-core/src/test/resources/org.apache.s4.deploy.s4.properties
index d74927f..1b03f23 100644
--- a/subprojects/s4-core/src/test/resources/org.apache.s4.deploy.s4.properties
+++ b/subprojects/s4-core/src/test/resources/org.apache.s4.deploy.s4.properties
@@ -8,4 +8,5 @@ cluster.zk_session_timeout = 10000
cluster.zk_connection_timeout = 10000
comm.module = org.apache.s4.deploy.TestModule
s4.logger_level = TRACE
-appsDir=/tmp/deploy-test
\ No newline at end of file
+appsDir=/tmp/deploy-test
+tcp.partition.queue_size=256
\ No newline at end of file