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/07/25 21:15:25 UTC
[6/13] git commit: Prefix all S4 config properties with "s4"
Prefix all S4 config properties with "s4"
Project: http://git-wip-us.apache.org/repos/asf/incubator-s4/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-s4/commit/cd026684
Tree: http://git-wip-us.apache.org/repos/asf/incubator-s4/tree/cd026684
Diff: http://git-wip-us.apache.org/repos/asf/incubator-s4/diff/cd026684
Branch: refs/heads/S4-88
Commit: cd026684cdba2b7765af7fd18e9dc0999cbfa227
Parents: c349777
Author: Matthieu Morel <mm...@apache.org>
Authored: Tue Jul 24 15:51:24 2012 +0200
Committer: Matthieu Morel <mm...@apache.org>
Committed: Tue Jul 24 19:09:34 2012 +0200
----------------------------------------------------------------------
.../java/org/apache/s4/comm/DefaultCommModule.java | 13 ++-
.../java/org/apache/s4/comm/tcp/TCPEmitter.java | 2 +-
.../java/org/apache/s4/comm/tcp/TCPListener.java | 2 +-
.../org/apache/s4/comm/tcp/TCPRemoteEmitter.java | 7 +-
.../apache/s4/comm/topology/AssignmentFromZK.java | 8 +-
.../org/apache/s4/comm/topology/ClusterFromZK.java | 8 +-
.../apache/s4/comm/topology/ClustersFromZK.java | 8 +-
.../org/apache/s4/comm/topology/RemoteStreams.java | 6 +-
.../src/main/resources/default.s4.comm.properties | 20 ++--
.../java/org/apache/s4/comm/DeliveryTestUtil.java | 4 +-
.../java/org/apache/s4/comm/tcp/TCPCommTest.java | 17 +---
.../org/apache/s4/comm/util/PartitionInfo.java | 13 +--
.../src/test/resources/udp.s4.comm.properties | 19 ++--
.../src/main/java/org/apache/s4/core/App.java | 2 +-
.../src/main/java/org/apache/s4/core/Server.java | 6 +-
.../org/apache/s4/core/adapter/AdapterApp.java | 2 +-
.../s4/deploy/DistributedDeploymentManager.java | 8 +-
.../src/main/resources/default.s4.core.properties | 1 -
.../test/java/org/apache/s4/deploy/TestModule.java | 85 ---------------
.../s4/deploy/prodcon/TestProducerConsumer.java | 31 +-----
.../s4/fixtures/AssignmentFromZKNoFailFast.java | 8 +-
.../s4/fixtures/ClusterFromZKNoFailFast.java | 8 +-
.../s4/fixtures/ClustersFromZKNoFailFast.java | 8 +-
.../org/apache/s4/fixtures/MockCommModule.java | 2 +-
.../src/main/resources/templates/newApp.README | 4 +-
25 files changed, 75 insertions(+), 217 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/cd026684/subprojects/s4-comm/src/main/java/org/apache/s4/comm/DefaultCommModule.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/DefaultCommModule.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/DefaultCommModule.java
index 49593ef..d7c8cee 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/DefaultCommModule.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/DefaultCommModule.java
@@ -98,17 +98,18 @@ public class DefaultCommModule extends AbstractModule {
try {
Class<? extends Emitter> emitterClass = (Class<? extends Emitter>) Class.forName(config
- .getString("comm.emitter.class"));
+ .getString("s4.comm.emitter.class"));
bind(Emitter.class).to(emitterClass);
// RemoteEmitter instances are created through a factory, depending on the topology. We inject the factory
Class<? extends RemoteEmitter> remoteEmitterClass = (Class<? extends RemoteEmitter>) Class.forName(config
- .getString("comm.emitter.remote.class"));
+ .getString("s4.comm.emitter.remote.class"));
install(new FactoryModuleBuilder().implement(RemoteEmitter.class, remoteEmitterClass).build(
RemoteEmitterFactory.class));
bind(RemoteEmitters.class);
- bind(Listener.class).to((Class<? extends Listener>) Class.forName(config.getString("comm.listener.class")));
+ bind(Listener.class).to(
+ (Class<? extends Listener>) Class.forName(config.getString("s4.comm.listener.class")));
} catch (ClassNotFoundException e) {
logger.error("Cannot find class implementation ", e);
@@ -128,14 +129,14 @@ public class DefaultCommModule extends AbstractModule {
Names.bindProperties(binder, ConfigurationConverter.getProperties(config));
if (clusterName != null) {
- if (config.containsKey("cluster.name")) {
+ if (config.containsKey("s4.cluster.name")) {
logger.warn(
"cluster [{}] passed as a parameter will not be used because an existing cluster.name parameter of value [{}] was found in the configuration file and will be used",
- clusterName, config.getProperty("cluster.name"));
+ clusterName, config.getProperty("s4.cluster.name"));
} else {
Names.bindProperties(binder, new HashMap<String, String>() {
{
- put("cluster.name", clusterName);
+ put("s4.cluster.name", clusterName);
}
});
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/cd026684/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tcp/TCPEmitter.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tcp/TCPEmitter.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tcp/TCPEmitter.java
index 8192cdc..44c9247 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tcp/TCPEmitter.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tcp/TCPEmitter.java
@@ -92,7 +92,7 @@ public class TCPEmitter implements Emitter, ClusterChangeListener {
SerializerDeserializer serDeser;
@Inject
- public TCPEmitter(Cluster topology, @Named("comm.timeout") int timeout) throws InterruptedException {
+ public TCPEmitter(Cluster topology, @Named("s4.comm.timeout") int timeout) throws InterruptedException {
this.nettyTimeout = timeout;
this.topology = topology;
this.lock = new ReentrantLock();
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/cd026684/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tcp/TCPListener.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tcp/TCPListener.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tcp/TCPListener.java
index d4d8b91..d495033 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tcp/TCPListener.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tcp/TCPListener.java
@@ -63,7 +63,7 @@ public class TCPListener implements Listener {
private final int nettyTimeout;
@Inject
- public TCPListener(Assignment assignment, @Named("comm.timeout") int timeout) {
+ public TCPListener(Assignment assignment, @Named("s4.comm.timeout") int timeout) {
// wait for an assignment
node = assignment.assignClusterNode();
nettyTimeout = timeout;
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/cd026684/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tcp/TCPRemoteEmitter.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tcp/TCPRemoteEmitter.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tcp/TCPRemoteEmitter.java
index a206906..a22767b 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tcp/TCPRemoteEmitter.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tcp/TCPRemoteEmitter.java
@@ -32,12 +32,11 @@ import com.google.inject.name.Named;
public class TCPRemoteEmitter extends TCPEmitter implements RemoteEmitter {
/**
- * Sends to remote subclusters. This is dynamically created, through an
- * injected factory, when new subclusters are discovered (as remote streams
- * outputs)
+ * Sends to remote subclusters. This is dynamically created, through an injected factory, when new subclusters are
+ * discovered (as remote streams outputs)
*/
@Inject
- public TCPRemoteEmitter(@Assisted Cluster topology, @Named("comm.timeout") int timeout)
+ public TCPRemoteEmitter(@Assisted Cluster topology, @Named("s4.comm.timeout") int timeout)
throws InterruptedException {
super(topology, timeout);
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/cd026684/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/AssignmentFromZK.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/AssignmentFromZK.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/AssignmentFromZK.java
index c3e2ffa..0b81854 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/AssignmentFromZK.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/AssignmentFromZK.java
@@ -90,10 +90,10 @@ public class AssignmentFromZK implements Assignment, IZkChildListener, IZkStateL
// TODO we currently have a single assignment per node (i.e. a node can only belong to 1 topology)
@Inject
- public AssignmentFromZK(@Named("cluster.name") String clusterName,
- @Named("cluster.zk_address") String zookeeperAddress,
- @Named("cluster.zk_session_timeout") int sessionTimeout,
- @Named("cluster.zk_connection_timeout") int connectionTimeout) throws Exception {
+ public AssignmentFromZK(@Named("s4.cluster.name") String clusterName,
+ @Named("s4.cluster.zk_address") String zookeeperAddress,
+ @Named("s4.cluster.zk_session_timeout") int sessionTimeout,
+ @Named("s4.cluster.zk_connection_timeout") int connectionTimeout) throws Exception {
this.clusterName = clusterName;
this.connectionTimeout = connectionTimeout;
taskPath = "/s4/clusters/" + clusterName + "/tasks";
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/cd026684/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ClusterFromZK.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ClusterFromZK.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ClusterFromZK.java
index 5635ff0..6bda6b4 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ClusterFromZK.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ClusterFromZK.java
@@ -59,10 +59,10 @@ public class ClusterFromZK implements Cluster, IZkChildListener, IZkDataListener
* only the local topology
*/
@Inject
- public ClusterFromZK(@Named("cluster.name") String clusterName,
- @Named("cluster.zk_address") String zookeeperAddress,
- @Named("cluster.zk_session_timeout") int sessionTimeout,
- @Named("cluster.zk_connection_timeout") int connectionTimeout) throws Exception {
+ public ClusterFromZK(@Named("s4.cluster.name") String clusterName,
+ @Named("s4.cluster.zk_address") String zookeeperAddress,
+ @Named("s4.cluster.zk_session_timeout") int sessionTimeout,
+ @Named("s4.cluster.zk_connection_timeout") int connectionTimeout) throws Exception {
this.clusterName = clusterName;
this.taskPath = "/s4/clusters/" + clusterName + "/tasks";
this.processPath = "/s4/clusters/" + clusterName + "/process";
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/cd026684/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ClustersFromZK.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ClustersFromZK.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ClustersFromZK.java
index 47fbacb..86f6bb5 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ClustersFromZK.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ClustersFromZK.java
@@ -51,10 +51,10 @@ public class ClustersFromZK implements Clusters, IZkStateListener {
private String clusterName;
@Inject
- public ClustersFromZK(@Named("cluster.name") String clusterName,
- @Named("cluster.zk_address") String zookeeperAddress,
- @Named("cluster.zk_session_timeout") int sessionTimeout,
- @Named("cluster.zk_connection_timeout") int connectionTimeout) throws Exception {
+ public ClustersFromZK(@Named("s4.cluster.name") String clusterName,
+ @Named("s4.cluster.zk_address") String zookeeperAddress,
+ @Named("s4.cluster.zk_session_timeout") int sessionTimeout,
+ @Named("s4.cluster.zk_connection_timeout") int connectionTimeout) throws Exception {
this.clusterName = clusterName;
this.connectionTimeout = connectionTimeout;
lock = new ReentrantLock();
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/cd026684/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/RemoteStreams.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/RemoteStreams.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/RemoteStreams.java
index c9d6e94..b1fe6b7 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/RemoteStreams.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/RemoteStreams.java
@@ -89,9 +89,9 @@ public class RemoteStreams implements IZkStateListener, IZkChildListener {
}
@Inject
- public RemoteStreams(@Named("cluster.zk_address") String zookeeperAddress,
- @Named("cluster.zk_session_timeout") int sessionTimeout,
- @Named("cluster.zk_connection_timeout") int connectionTimeout) throws Exception {
+ public RemoteStreams(@Named("s4.cluster.zk_address") String zookeeperAddress,
+ @Named("s4.cluster.zk_session_timeout") int sessionTimeout,
+ @Named("s4.cluster.zk_connection_timeout") int connectionTimeout) throws Exception {
lock = new ReentrantLock();
zkClient = new ZkClient(zookeeperAddress, sessionTimeout, connectionTimeout);
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/cd026684/subprojects/s4-comm/src/main/resources/default.s4.comm.properties
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/main/resources/default.s4.comm.properties b/subprojects/s4-comm/src/main/resources/default.s4.comm.properties
index 0c5bbfd..36417b4 100644
--- a/subprojects/s4-comm/src/main/resources/default.s4.comm.properties
+++ b/subprojects/s4-comm/src/main/resources/default.s4.comm.properties
@@ -1,12 +1,8 @@
-comm.queue_emmiter_size = 8000
-comm.queue_listener_size = 8000
-comm.emitter.class=org.apache.s4.comm.tcp.TCPEmitter
-comm.emitter.remote.class=org.apache.s4.comm.tcp.TCPRemoteEmitter
-comm.listener.class=org.apache.s4.comm.tcp.TCPListener
-comm.retries=10
-comm.retry_delay=10
-comm.timeout=1000
-tcp.partition.queue_size=256
-cluster.zk_address = localhost:2181
-cluster.zk_session_timeout = 10000
-cluster.zk_connection_timeout = 10000
\ No newline at end of file
+s4.comm.emitter.class=org.apache.s4.comm.tcp.TCPEmitter
+s4.comm.emitter.remote.class=org.apache.s4.comm.tcp.TCPRemoteEmitter
+s4.comm.listener.class=org.apache.s4.comm.tcp.TCPListener
+# I/O channel connection timeout, when applicable (e.g. used by netty)
+s4.comm.timeout=1000
+s4.cluster.zk_address = localhost:2181
+s4.cluster.zk_session_timeout = 10000
+s4.cluster.zk_connection_timeout = 10000
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/cd026684/subprojects/s4-comm/src/test/java/org/apache/s4/comm/DeliveryTestUtil.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/test/java/org/apache/s4/comm/DeliveryTestUtil.java b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/DeliveryTestUtil.java
index 45361d9..17340b8 100644
--- a/subprojects/s4-comm/src/test/java/org/apache/s4/comm/DeliveryTestUtil.java
+++ b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/DeliveryTestUtil.java
@@ -49,8 +49,8 @@ public class DeliveryTestUtil {
private final int messagesExpected;
@Inject
- public DeliveryTestUtil(Emitter emitter, Listener listener, @Named("emitter.send.interval") int interval,
- @Named("emitter.send.numMessages") int numMessages, @Named("listener.recv.sleepCount") int sleepCount) {
+ public DeliveryTestUtil(Emitter emitter, Listener listener, @Named("s4.emitter.send.interval") int interval,
+ @Named("s4.emitter.send.numMessages") int numMessages, @Named("s4.listener.recv.sleepCount") int sleepCount) {
this.emitter = emitter;
this.listener = listener;
this.interval = interval;
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/cd026684/subprojects/s4-comm/src/test/java/org/apache/s4/comm/tcp/TCPCommTest.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/test/java/org/apache/s4/comm/tcp/TCPCommTest.java b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/tcp/TCPCommTest.java
index 0cb26f3..c9153ce 100644
--- a/subprojects/s4-comm/src/test/java/org/apache/s4/comm/tcp/TCPCommTest.java
+++ b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/tcp/TCPCommTest.java
@@ -28,10 +28,8 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.common.io.Resources;
-import com.google.inject.AbstractModule;
import com.google.inject.Guice;
import com.google.inject.Injector;
-import com.google.inject.name.Names;
public abstract class TCPCommTest extends ProtocolTestUtil {
@@ -50,26 +48,13 @@ public abstract class TCPCommTest extends ProtocolTestUtil {
public Injector newInjector() {
try {
return Guice.createInjector(new DefaultCommModule(Resources.getResource("default.s4.comm.properties")
- .openStream(), CLUSTER_NAME), new TCPCommTestModule());
+ .openStream(), CLUSTER_NAME));
} catch (IOException e) {
Assert.fail();
return null;
}
}
- class TCPCommTestModule extends AbstractModule {
- TCPCommTestModule() {
-
- }
-
- @Override
- protected void configure() {
- bind(Integer.class).annotatedWith(Names.named("emitter.send.interval")).toInstance(100);
- bind(Integer.class).annotatedWith(Names.named("emitter.send.numMessages")).toInstance(200);
- bind(Integer.class).annotatedWith(Names.named("listener.recv.sleepCount")).toInstance(10);
- }
- }
-
@Override
public void testDelivery() throws InterruptedException {
startThreads();
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/cd026684/subprojects/s4-comm/src/test/java/org/apache/s4/comm/util/PartitionInfo.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/test/java/org/apache/s4/comm/util/PartitionInfo.java b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/util/PartitionInfo.java
index 5e9df6e..97cc973 100644
--- a/subprojects/s4-comm/src/test/java/org/apache/s4/comm/util/PartitionInfo.java
+++ b/subprojects/s4-comm/src/test/java/org/apache/s4/comm/util/PartitionInfo.java
@@ -30,7 +30,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.google.inject.Inject;
-import com.google.inject.name.Named;
/**
* Test util for communication protocols.
@@ -51,9 +50,9 @@ public class PartitionInfo {
public SendThread sendThread;
public ReceiveThread receiveThread;
- private final int numRetries;
- private final int retryDelayMs;
- private int numMessages;
+ private final int numRetries = 10;
+ private final int retryDelayMs = 10;
+ private int numMessages = 100;
private int partitionId;
private ProtocolTestUtil ptu;
@@ -61,17 +60,13 @@ public class PartitionInfo {
SerializerDeserializer serDeser;
@Inject
- public PartitionInfo(Emitter emitter, Listener listener, @Named("comm.retries") int retries,
- @Named("comm.retry_delay") int retryDelay, @Named("emitter.send.numMessages") int numMessages) {
+ public PartitionInfo(Emitter emitter, Listener listener) {
this.emitter = emitter;
this.listener = listener;
this.partitionId = this.listener.getPartitionId();
logger.debug("# Partitions = {}; Current partition = {}", this.emitter.getPartitionCount(),
this.listener.getPartitionId());
- this.numRetries = retries;
- this.retryDelayMs = retryDelay;
- this.numMessages = numMessages;
// this.messagesExpected = numMessages * this.emitter.getPartitionCount();
this.sendThread = new SendThread();
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/cd026684/subprojects/s4-comm/src/test/resources/udp.s4.comm.properties
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/test/resources/udp.s4.comm.properties b/subprojects/s4-comm/src/test/resources/udp.s4.comm.properties
index a5367c7..7b525bd 100644
--- a/subprojects/s4-comm/src/test/resources/udp.s4.comm.properties
+++ b/subprojects/s4-comm/src/test/resources/udp.s4.comm.properties
@@ -1,12 +1,7 @@
-comm.queue_emmiter_size = 8000
-comm.queue_listener_size = 8000
-comm.emitter.class = org.apache.s4.comm.udp.UDPEmitter
-comm.emitter.remote.class = org.apache.s4.comm.udp.UDPRemoteEmitter
-comm.listener.class = org.apache.s4.comm.udp.UDPListener
-cluster.name = cluster1
-cluster.zk_address = localhost:2181
-cluster.zk_session_timeout = 10000
-cluster.zk_connection_timeout = 10000
-comm.retries=10
-comm.retry_delay=10
-comm.timeout=1000
\ No newline at end of file
+s4.comm.emitter.class = org.apache.s4.comm.udp.UDPEmitter
+s4.comm.emitter.remote.class = org.apache.s4.comm.udp.UDPRemoteEmitter
+s4.comm.listener.class = org.apache.s4.comm.udp.UDPListener
+s4.cluster.name = cluster1
+s4.cluster.zk_address = localhost:2181
+s4.cluster.zk_session_timeout = 10000
+s4.cluster.zk_connection_timeout = 10000
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/cd026684/subprojects/s4-core/src/main/java/org/apache/s4/core/App.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/App.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/App.java
index 7c6dc9b..936d225 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/core/App.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/App.java
@@ -77,7 +77,7 @@ public abstract class App {
RemoteStreams remoteStreams;
@Inject
- @Named("cluster.name")
+ @Named("s4.cluster.name")
String clusterName;
// default is NoOpCheckpointingFramework
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/cd026684/subprojects/s4-core/src/main/java/org/apache/s4/core/Server.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/Server.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/Server.java
index 32aa9a1..02e521a 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/core/Server.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/Server.java
@@ -66,9 +66,9 @@ public class Server {
*/
@Inject
public Server(String commModuleName, @Named("s4.logger_level") String logLevel,
- @Named("cluster.name") String clusterName, @Named("cluster.zk_address") String zookeeperAddress,
- @Named("cluster.zk_session_timeout") int sessionTimeout,
- @Named("cluster.zk_connection_timeout") int connectionTimeout) {
+ @Named("s4.cluster.name") String clusterName, @Named("s4.cluster.zk_address") String zookeeperAddress,
+ @Named("s4.cluster.zk_session_timeout") int sessionTimeout,
+ @Named("s4.cluster.zk_connection_timeout") int connectionTimeout) {
this.logLevel = logLevel;
zkClient = new ZkClient(zookeeperAddress, sessionTimeout, connectionTimeout);
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/cd026684/subprojects/s4-core/src/main/java/org/apache/s4/core/adapter/AdapterApp.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/adapter/AdapterApp.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/adapter/AdapterApp.java
index 6466b29..f0c45a9 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/core/adapter/AdapterApp.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/adapter/AdapterApp.java
@@ -33,7 +33,7 @@ import com.google.inject.name.Named;
public abstract class AdapterApp extends App {
@Inject
- @Named(value = "adapter.output.stream")
+ @Named(value = "s4.adapter.output.stream")
String outputStreamName;
private RemoteStream remoteStream;
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/cd026684/subprojects/s4-core/src/main/java/org/apache/s4/deploy/DistributedDeploymentManager.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/deploy/DistributedDeploymentManager.java b/subprojects/s4-core/src/main/java/org/apache/s4/deploy/DistributedDeploymentManager.java
index 156e157..9fc5c53 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/deploy/DistributedDeploymentManager.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/deploy/DistributedDeploymentManager.java
@@ -79,10 +79,10 @@ public class DistributedDeploymentManager implements DeploymentManager {
boolean deployed = false;
@Inject
- public DistributedDeploymentManager(@Named("cluster.name") String clusterName,
- @Named("cluster.zk_address") String zookeeperAddress,
- @Named("cluster.zk_session_timeout") int sessionTimeout,
- @Named("cluster.zk_connection_timeout") int connectionTimeout, Server server) {
+ public DistributedDeploymentManager(@Named("s4.cluster.name") String clusterName,
+ @Named("s4.cluster.zk_address") String zookeeperAddress,
+ @Named("s4.cluster.zk_session_timeout") int sessionTimeout,
+ @Named("s4.cluster.zk_connection_timeout") int connectionTimeout, Server server) {
this.clusterName = clusterName;
this.server = server;
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/cd026684/subprojects/s4-core/src/main/resources/default.s4.core.properties
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/resources/default.s4.core.properties b/subprojects/s4-core/src/main/resources/default.s4.core.properties
index a078f75..cb5e20a 100644
--- a/subprojects/s4-core/src/main/resources/default.s4.core.properties
+++ b/subprojects/s4-core/src/main/resources/default.s4.core.properties
@@ -1,2 +1 @@
s4.logger_level = DEBUG
-appsDir=/tmp/deploy-test
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/cd026684/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
deleted file mode 100644
index 1f73d38..0000000
--- a/subprojects/s4-core/src/test/java/org/apache/s4/deploy/TestModule.java
+++ /dev/null
@@ -1,85 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.s4.deploy;
-
-import java.io.InputStream;
-
-import org.apache.commons.configuration.ConfigurationConverter;
-import org.apache.commons.configuration.ConfigurationException;
-import org.apache.commons.configuration.ConfigurationUtils;
-import org.apache.commons.configuration.PropertiesConfiguration;
-import org.apache.s4.base.Emitter;
-import org.apache.s4.base.Hasher;
-import org.apache.s4.base.Listener;
-import org.apache.s4.base.SerializerDeserializer;
-import org.apache.s4.comm.DefaultHasher;
-import org.apache.s4.comm.serialize.KryoSerDeser;
-import org.apache.s4.comm.tcp.TCPEmitter;
-import org.apache.s4.comm.tcp.TCPListener;
-import org.apache.s4.comm.topology.Assignment;
-import org.apache.s4.comm.topology.AssignmentFromZK;
-import org.apache.s4.comm.topology.Cluster;
-import org.apache.s4.comm.topology.ClusterFromZK;
-
-import com.google.inject.AbstractModule;
-import com.google.inject.Binder;
-import com.google.inject.name.Names;
-
-public class TestModule extends AbstractModule {
-
- private PropertiesConfiguration config;
-
- private void loadProperties(Binder binder) {
-
- try {
- InputStream is = this.getClass().getResourceAsStream("/org.apache.s4.deploy.s4.properties");
- config = new PropertiesConfiguration();
- config.load(is);
- System.out.println(ConfigurationUtils.toString(config));
- // TODO - validate properties.
-
- /* Make all properties injectable. Do we need this? */
- Names.bindProperties(binder, ConfigurationConverter.getProperties(config));
- } catch (ConfigurationException e) {
- binder.addError(e);
- e.printStackTrace();
- }
- }
-
- @Override
- protected void configure() {
- if (config == null) {
- loadProperties(binder());
- }
- bind(Cluster.class);
- bind(Hasher.class).to(DefaultHasher.class);
- bind(SerializerDeserializer.class).to(KryoSerDeser.class);
- bind(Assignment.class).to(AssignmentFromZK.class);
- bind(Cluster.class).to(ClusterFromZK.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/cd026684/subprojects/s4-core/src/test/java/org/apache/s4/deploy/prodcon/TestProducerConsumer.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/test/java/org/apache/s4/deploy/prodcon/TestProducerConsumer.java b/subprojects/s4-core/src/test/java/org/apache/s4/deploy/prodcon/TestProducerConsumer.java
index 4a6840a..442bbcd 100644
--- a/subprojects/s4-core/src/test/java/org/apache/s4/deploy/prodcon/TestProducerConsumer.java
+++ b/subprojects/s4-core/src/test/java/org/apache/s4/deploy/prodcon/TestProducerConsumer.java
@@ -46,7 +46,6 @@ import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;
-import com.google.common.io.ByteStreams;
import com.google.common.io.Files;
import com.google.common.io.Resources;
import com.sun.net.httpserver.HttpServer;
@@ -79,20 +78,6 @@ public class TestProducerConsumer {
}
@Before
- public void cleanLocalAppsDir() throws ConfigurationException, IOException {
- PropertiesConfiguration config = loadConfig();
-
- if (!new File(config.getString("appsDir")).exists()) {
- Assert.assertTrue(new File(config.getString("appsDir")).mkdirs());
- } else {
- if (!config.getString("appsDir").startsWith("/tmp")) {
- Assert.fail("apps dir should a subdir of /tmp for safety");
- }
- CommTestUtils.deleteDirectoryContents(new File(config.getString("appsDir")));
- }
- }
-
- @Before
public void prepare() throws Exception {
CommTestUtils.cleanupTmpDirs();
zookeeperServerConnectionFactory = CommTestUtils.startZookeeperServer();
@@ -122,18 +107,10 @@ public class TestProducerConsumer {
@Test
public void testInitialDeploymentFromFileSystem() throws Exception {
- File producerS4R = new File(loadConfig().getString("appsDir") + File.separator + "producer"
- + System.currentTimeMillis() + ".s4r");
- System.out.println(tmpAppsDir.getAbsolutePath());
- Assert.assertTrue(ByteStreams.copy(Files.newInputStreamSupplier(new File(tmpAppsDir.getAbsolutePath()
- + "/producer-app-0.0.0-SNAPSHOT.s4r")), Files.newOutputStreamSupplier(producerS4R)) > 0);
+ File producerS4R = new File(tmpAppsDir, "producer-app-0.0.0-SNAPSHOT.s4r");
String uriProducer = producerS4R.toURI().toString();
- File consumerS4R = new File(loadConfig().getString("appsDir") + File.separator + "consumer"
- + System.currentTimeMillis() + ".s4r");
- Assert.assertTrue(ByteStreams.copy(Files.newInputStreamSupplier(new File(tmpAppsDir.getAbsolutePath()
- + "/consumer-app-0.0.0-SNAPSHOT.s4r")), Files.newOutputStreamSupplier(consumerS4R)) > 0);
-
+ File consumerS4R = new File(tmpAppsDir, "consumer-app-0.0.0-SNAPSHOT.s4r");
String uriConsumer = consumerS4R.toURI().toString();
initializeS4Node();
@@ -178,10 +155,6 @@ public class TestProducerConsumer {
}
private void initializeS4Node() throws ConfigurationException, IOException, InterruptedException, KeeperException {
- // 0. package s4 app
- // TODO this is currently done offline, and the app contains the TestApp class copied from the one in the
- // current package .
-
// 1. start s4 nodes. Check that no app is deployed.
TaskSetup taskSetup = new TaskSetup("localhost:" + CommTestUtils.ZK_PORT);
taskSetup.clean("s4");
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/cd026684/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/AssignmentFromZKNoFailFast.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/AssignmentFromZKNoFailFast.java b/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/AssignmentFromZKNoFailFast.java
index 144e37e..e8b7e87 100644
--- a/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/AssignmentFromZKNoFailFast.java
+++ b/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/AssignmentFromZKNoFailFast.java
@@ -27,10 +27,10 @@ import com.google.inject.name.Named;
public class AssignmentFromZKNoFailFast extends AssignmentFromZK {
@Inject
- public AssignmentFromZKNoFailFast(@Named("cluster.name") String clusterName,
- @Named("cluster.zk_address") String zookeeperAddress,
- @Named("cluster.zk_session_timeout") int sessionTimeout,
- @Named("cluster.zk_connection_timeout") int connectionTimeout) throws Exception {
+ public AssignmentFromZKNoFailFast(@Named("s4.cluster.name") String clusterName,
+ @Named("s4.cluster.zk_address") String zookeeperAddress,
+ @Named("s4.cluster.zk_session_timeout") int sessionTimeout,
+ @Named("s4.cluster.zk_connection_timeout") int connectionTimeout) throws Exception {
super(clusterName, zookeeperAddress, sessionTimeout, connectionTimeout);
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/cd026684/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/ClusterFromZKNoFailFast.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/ClusterFromZKNoFailFast.java b/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/ClusterFromZKNoFailFast.java
index 272117f..025fb75 100644
--- a/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/ClusterFromZKNoFailFast.java
+++ b/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/ClusterFromZKNoFailFast.java
@@ -27,10 +27,10 @@ import com.google.inject.name.Named;
public class ClusterFromZKNoFailFast extends ClusterFromZK {
@Inject
- public ClusterFromZKNoFailFast(@Named("cluster.name") String clusterName,
- @Named("cluster.zk_address") String zookeeperAddress,
- @Named("cluster.zk_session_timeout") int sessionTimeout,
- @Named("cluster.zk_connection_timeout") int connectionTimeout) throws Exception {
+ public ClusterFromZKNoFailFast(@Named("s4.cluster.name") String clusterName,
+ @Named("s4.cluster.zk_address") String zookeeperAddress,
+ @Named("s4.cluster.zk_session_timeout") int sessionTimeout,
+ @Named("s4.cluster.zk_connection_timeout") int connectionTimeout) throws Exception {
super(clusterName, zookeeperAddress, sessionTimeout, connectionTimeout);
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/cd026684/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/ClustersFromZKNoFailFast.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/ClustersFromZKNoFailFast.java b/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/ClustersFromZKNoFailFast.java
index cf72048..4ea5644 100644
--- a/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/ClustersFromZKNoFailFast.java
+++ b/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/ClustersFromZKNoFailFast.java
@@ -27,10 +27,10 @@ import com.google.inject.name.Named;
public class ClustersFromZKNoFailFast extends ClustersFromZK {
@Inject
- public ClustersFromZKNoFailFast(@Named("cluster.name") String clusterName,
- @Named("cluster.zk_address") String zookeeperAddress,
- @Named("cluster.zk_session_timeout") int sessionTimeout,
- @Named("cluster.zk_connection_timeout") int connectionTimeout) throws Exception {
+ public ClustersFromZKNoFailFast(@Named("s4.cluster.name") String clusterName,
+ @Named("s4.cluster.zk_address") String zookeeperAddress,
+ @Named("s4.cluster.zk_session_timeout") int sessionTimeout,
+ @Named("s4.cluster.zk_connection_timeout") int connectionTimeout) throws Exception {
super(clusterName, zookeeperAddress, sessionTimeout, connectionTimeout);
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/cd026684/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/MockCommModule.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/MockCommModule.java b/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/MockCommModule.java
index 57126af..75143c1 100644
--- a/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/MockCommModule.java
+++ b/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/MockCommModule.java
@@ -59,7 +59,7 @@ public class MockCommModule extends AbstractModule {
Assignment mockedAssignment = Mockito.mock(Assignment.class);
Mockito.when(mockedAssignment.assignClusterNode()).thenReturn(new ClusterNode(0, 0, "machine", "Task-0"));
bind(Assignment.class).toInstance(mockedAssignment);
- Names.bindProperties(binder(), ImmutableMap.of("cluster.name", "testCluster"));
+ Names.bindProperties(binder(), ImmutableMap.of("s4.cluster.name", "testCluster"));
}
}
http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/cd026684/subprojects/s4-tools/src/main/resources/templates/newApp.README
----------------------------------------------------------------------
diff --git a/subprojects/s4-tools/src/main/resources/templates/newApp.README b/subprojects/s4-tools/src/main/resources/templates/newApp.README
index a174e31..2ce7c36 100644
--- a/subprojects/s4-tools/src/main/resources/templates/newApp.README
+++ b/subprojects/s4-tools/src/main/resources/templates/newApp.README
@@ -20,7 +20,7 @@
To execute the application in a new S4 cluster:
- 1. start a ZooKeeper instance "./s4 zkServer"
+ 1. start a ZooKeeper instance "./s4 zkServer" (./s4 zkServer -help provides a list of options)
2. define a logical cluster for your application "./s4 newCluster -cluster=<nameOfTheCluster> -nbTasks=<number of partitions> -flp=<a port number for the first node, other nodes use an increment on this initial port>"
3. start a node and attach it to the cluster "./s4 node -cluster=<nameOfTheCluster>"
4. deploy the application "./s4 deploy -cluster=<nameOfTheCluster>"
@@ -31,5 +31,5 @@
If you want to use a simple adapter process, listening to an external source, converting incoming data into S4 events, and sending that to S4 apps, you can define
your own app that extends the AdapterApp class. Then to start it (for instance):
- - ./s4 adapter -cluster=c1 -appClass=org.apache.s4.example.twitter.TwitterInputAdapter -p={adapter.output.stream=stream1}
+ - ./s4 adapter -cluster=c1 -appClass=org.apache.s4.example.twitter.TwitterInputAdapter -p=s4.adapter.output.stream=stream1