incubator-s4-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From kisho...@apache.org
Subject [2/2] git commit: interim checkin. Cant get the bindings working
Date Mon, 04 Feb 2013 04:40:11 GMT
Updated Branches:
  refs/heads/S4-110-new [created] 55f8c2157


interim checkin. Cant get the bindings working


Project: http://git-wip-us.apache.org/repos/asf/incubator-s4/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-s4/commit/55f8c215
Tree: http://git-wip-us.apache.org/repos/asf/incubator-s4/tree/55f8c215
Diff: http://git-wip-us.apache.org/repos/asf/incubator-s4/diff/55f8c215

Branch: refs/heads/S4-110-new
Commit: 55f8c2157d267f22bddea56ecf6ed2c3540ea3d0
Parents: 6cd8742
Author: Kishore Gopalakrishna <g.kishore@gmail.com>
Authored: Sun Feb 3 20:38:13 2013 -0800
Committer: Kishore Gopalakrishna <g.kishore@gmail.com>
Committed: Sun Feb 3 20:38:13 2013 -0800

----------------------------------------------------------------------
 build.gradle                                       |   15 +
 .../src/main/java/org/apache/s4/base/Emitter.java  |    2 +
 .../org/apache/s4/comm/HelixBasedCommModule.java   |  147 +++++
 .../org/apache/s4/comm/helix/S4StateModel.java     |   54 ++
 .../s4/comm/helix/TaskStateModelFactory.java       |   12 +
 .../java/org/apache/s4/comm/tcp/TCPEmitter.java    |    5 +
 .../s4/comm/topology/AssignmentFromHelix.java      |  143 +++++
 .../java/org/apache/s4/comm/topology/Cluster.java  |   11 +
 .../apache/s4/comm/topology/ClusterFromHelix.java  |  190 ++++++
 .../org/apache/s4/comm/topology/ClusterFromZK.java |   10 +
 .../apache/s4/comm/topology/ClustersFromHelix.java |   64 ++
 .../apache/s4/comm/topology/PhysicalCluster.java   |    8 +
 .../java/org/apache/s4/comm/udp/UDPEmitter.java    |    5 +
 .../main/java/org/apache/s4/core/BaseModule.java   |   50 ++-
 .../main/java/org/apache/s4/core/Bootstrap.java    |    7 +
 .../java/org/apache/s4/core/DefaultCoreModule.java |    8 +-
 .../org/apache/s4/core/HelixBasedCoreModule.java   |  134 ++++
 .../main/java/org/apache/s4/core/S4Bootstrap.java  |    2 +-
 .../java/org/apache/s4/core/S4HelixBootstrap.java  |  139 +++++
 .../src/main/java/org/apache/s4/core/S4Node.java   |    6 +-
 .../java/org/apache/s4/deploy/AppStateModel.java   |   73 +++
 .../org/apache/s4/deploy/AppStateModelFactory.java |   25 +
 .../org/apache/s4/deploy/DeploymentManager.java    |    9 +-
 .../java/org/apache/s4/deploy/DeploymentUtils.java |   65 ++
 .../s4/deploy/DistributedDeploymentManager.java    |   72 +--
 .../s4/deploy/HelixBasedDeploymentManager.java     |   42 ++
 .../apache/s4/deploy/NoOpDeploymentManager.java    |   14 +
 .../core/moduleloader/ModuleLoaderTestUtils.java   |    2 +-
 .../java/org/apache/s4/fixtures/CoreTestUtils.java |    2 +-
 .../org/apache/s4/wordcount/WordCountTest.java     |    2 +-
 .../src/main/java/org/apache/s4/tools/Tools.java   |   18 +-
 .../java/org/apache/s4/tools/helix/AddNodes.java   |   90 +++
 .../org/apache/s4/tools/helix/CreateCluster.java   |   99 +++
 .../java/org/apache/s4/tools/helix/CreateTask.java |   91 +++
 .../java/org/apache/s4/tools/helix/DeployApp.java  |   85 +++
 .../apache/s4/tools/helix/GenericEventAdapter.java |   72 +++
 .../org/apache/s4/tools/helix/RebalanceTask.java   |   61 ++
 .../java/org/apache/s4/tools/helix/S4Status.java   |  468 +++++++++++++++
 38 files changed, 2231 insertions(+), 71 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/build.gradle
----------------------------------------------------------------------
diff --git a/build.gradle b/build.gradle
index b36a770..828ba30 100644
--- a/build.gradle
+++ b/build.gradle
@@ -40,6 +40,7 @@ allprojects {
         maven { url 'http://google-gson.googlecode.com/svn/mavenrepo' }
         maven { url 'https://repo.springsource.org/libs-release' }
         maven { url 'http://repo.gradle.org/gradle/libs-releases-local' }
+        maven { url 'http://maven.restlet.org' }
 
         /* Add lib dir as a repo. Some jar files that are not available
          in a public repo are distributed in the lib dir. */
@@ -71,6 +72,10 @@ project.ext["libraries"] = [
     commons_lang:       'commons-lang:commons-lang:2.6',
     commons_logging:    'commons-logging:commons-logging:1.1.1',
     commons_collections:'commons-collections:commons-collections:3.2.1',
+    commons_cli:        'commons-cli:commons-cli:1.2',
+    commons_math:       'org.apache.commons:commons-math:2.1',
+    restlet_neo:        'com.noelios.restlet:com.noelios.restlet:1.1.10',
+    restlet:            'org.restlet:org.restlet:1.1.10',
     objenesis:          'org.objenesis:objenesis:1.2',
     slf4j:              'org.slf4j:slf4j-api:1.6.1',
     log4j:              'log4j:log4j:1.2.15',
@@ -80,6 +85,9 @@ project.ext["libraries"] = [
     jcip:               'net.jcip:jcip-annotations:1.0',
     junit:              'junit:junit:4.10',
     zkclient:           'com.github.sgroschupf:zkclient:0.1',
+    helix:              'org.apache.helix:helix-core:0.6.0-incubating',
+    jackson_core_asl:   'org.codehaus.jackson:jackson-core-asl:1.8.5',  
+    jackson_mapper_asl: 'org.codehaus.jackson:jackson-mapper-asl:1.8.5',  
     diezel:             'net.ericaro:diezel-maven-plugin:1.0.0-beta-4',
     jcommander:         'com.beust:jcommander:1.25',
     asm:                'org.ow2.asm:asm:4.0',
@@ -130,11 +138,18 @@ subprojects {
         compile( libraries.commons_io )
         compile ( libraries.commons_lang )
         runtime(libraries.commons_collections)
+        runtime(libraries.commons_cli)
+        runtime(libraries.commons_math)
 
         /* Misc. */
         compile( libraries.jcip )
         compile( libraries.zk )
         compile( libraries.metrics )
+        compile(libraries.helix)
+        compile(libraries.jackson_core_asl)
+        compile(libraries.jackson_mapper_asl)
+        runtime(libraries.restlet)
+        runtime(libraries.restlet_neo)
 
         /* Testing. */
         testCompile( libraries.junit )

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/subprojects/s4-base/src/main/java/org/apache/s4/base/Emitter.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-base/src/main/java/org/apache/s4/base/Emitter.java b/subprojects/s4-base/src/main/java/org/apache/s4/base/Emitter.java
index 1f74381..e2ee412 100644
--- a/subprojects/s4-base/src/main/java/org/apache/s4/base/Emitter.java
+++ b/subprojects/s4-base/src/main/java/org/apache/s4/base/Emitter.java
@@ -40,6 +40,8 @@ public interface Emitter {
     boolean send(int partitionId, ByteBuffer message) throws InterruptedException;
 
     int getPartitionCount();
+    
+    int getPartitionCount(String stream);
 
     void close();
 }

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/subprojects/s4-comm/src/main/java/org/apache/s4/comm/HelixBasedCommModule.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/HelixBasedCommModule.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/HelixBasedCommModule.java
new file mode 100644
index 0000000..980f9bd
--- /dev/null
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/HelixBasedCommModule.java
@@ -0,0 +1,147 @@
+package org.apache.s4.comm;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.HashMap;
+
+import org.apache.commons.configuration.ConfigurationConverter;
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.commons.configuration.PropertiesConfiguration;
+import org.apache.helix.participant.statemachine.StateModelFactory;
+import org.apache.s4.base.Emitter;
+import org.apache.s4.base.Hasher;
+import org.apache.s4.base.Listener;
+import org.apache.s4.base.RemoteEmitter;
+import org.apache.s4.base.SerializerDeserializer;
+import org.apache.s4.comm.helix.TaskStateModelFactory;
+import org.apache.s4.comm.serialize.KryoSerDeser;
+import org.apache.s4.comm.serialize.SerializerDeserializerFactory;
+import org.apache.s4.comm.staging.BlockingDeserializerExecutorFactory;
+import org.apache.s4.comm.tcp.DefaultRemoteEmitters;
+import org.apache.s4.comm.tcp.RemoteEmitters;
+import org.apache.s4.comm.topology.Assignment;
+import org.apache.s4.comm.topology.AssignmentFromHelix;
+import org.apache.s4.comm.topology.AssignmentFromZK;
+import org.apache.s4.comm.topology.Cluster;
+import org.apache.s4.comm.topology.ClusterFromHelix;
+import org.apache.s4.comm.topology.ClusterFromZK;
+import org.apache.s4.comm.topology.Clusters;
+import org.apache.s4.comm.topology.ClustersFromHelix;
+import org.apache.s4.comm.topology.ClustersFromZK;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.inject.AbstractModule;
+import com.google.inject.Binder;
+import com.google.inject.Scopes;
+import com.google.inject.assistedinject.FactoryModuleBuilder;
+import com.google.inject.name.Names;
+
+public class HelixBasedCommModule extends AbstractModule {
+
+    private static Logger logger = LoggerFactory.getLogger(DefaultCommModule.class);
+    InputStream commConfigInputStream;
+    private PropertiesConfiguration config;
+    String clusterName;
+
+    /**
+     * 
+     * @param commConfigInputStream
+     *            input stream from a configuration file
+     * @param clusterName
+     *            the name of the cluster to which the current node belongs. If specified in the configuration file,
+     *            this parameter will be ignored.
+     */
+    public HelixBasedCommModule(InputStream commConfigInputStream, String clusterName) {
+        super();
+        this.commConfigInputStream = commConfigInputStream;
+        this.clusterName = clusterName;
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override
+    protected void configure() {
+        if (config == null) {
+            loadProperties(binder());
+        }
+        if (commConfigInputStream != null) {
+            try {
+                commConfigInputStream.close();
+            } catch (IOException ignored) {
+            }
+        }
+
+        /* The hashing function to map keys top partitions. */
+        bind(Hasher.class).to(DefaultHasher.class);
+        /* Use Kryo to serialize events. */
+        install(new FactoryModuleBuilder().implement(SerializerDeserializer.class, KryoSerDeser.class).build(
+                SerializerDeserializerFactory.class));
+
+        // a node holds a single partition assignment
+        // ==> Assignment and Cluster are singletons so they can be shared between comm layer and app.
+        bind(StateModelFactory.class).annotatedWith(Names.named("s4.task.statemodelfactory")).to(
+                TaskStateModelFactory.class);
+        
+        bind(Assignment.class).to(AssignmentFromHelix.class);
+        
+        bind(Cluster.class).to(ClusterFromHelix.class);
+
+        bind(Clusters.class).to(ClustersFromHelix.class);
+
+        bind(RemoteEmitters.class).to(DefaultRemoteEmitters.class).in(Scopes.SINGLETON);
+
+        bind(DeserializerExecutorFactory.class).to(BlockingDeserializerExecutorFactory.class);
+        
+        try {
+            Class<? extends Emitter> emitterClass = (Class<? extends Emitter>) Class.forName(config
+                    .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("s4.comm.emitter.remote.class"));
+            install(new FactoryModuleBuilder().implement(RemoteEmitter.class, remoteEmitterClass).build(
+                    RemoteEmitterFactory.class));
+            bind(RemoteEmitters.class).to(DefaultRemoteEmitters.class).in(Scopes.SINGLETON);
+
+//            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);
+        }
+
+    }
+
+    @SuppressWarnings("serial")
+    private void loadProperties(Binder binder) {
+        try {
+            config = new PropertiesConfiguration();
+            config.load(commConfigInputStream);
+
+            // TODO - validate properties.
+
+            /* Make all properties injectable. Do we need this? */
+            Names.bindProperties(binder, ConfigurationConverter.getProperties(config));
+
+            if (clusterName != null) {
+                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("s4.cluster.name"));
+                } else {
+                    Names.bindProperties(binder, new HashMap<String, String>() {
+                        {
+                            put("s4.cluster.name", clusterName);
+                        }
+                    });
+                }
+            }
+
+        } catch (ConfigurationException e) {
+            binder.addError(e);
+            e.printStackTrace();
+        }
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/subprojects/s4-comm/src/main/java/org/apache/s4/comm/helix/S4StateModel.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/helix/S4StateModel.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/helix/S4StateModel.java
new file mode 100644
index 0000000..ea35bd6
--- /dev/null
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/helix/S4StateModel.java
@@ -0,0 +1,54 @@
+package org.apache.s4.comm.helix;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.helix.NotificationContext;
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.statemachine.StateModel;
+import org.apache.helix.participant.statemachine.StateModelInfo;
+import org.apache.helix.participant.statemachine.Transition;
+
+@StateModelInfo(states = { "LEADER,STANDBY" }, initialState = "OFFLINE")
+public class S4StateModel extends StateModel {
+    private static Logger logger = LoggerFactory.getLogger(S4StateModel.class);
+
+    private final String streamName;
+    private final String partitionId;
+
+    public S4StateModel(String partitionName) {
+        String[] parts = partitionName.split("_");
+        this.streamName = parts[0];
+        this.partitionId = parts[1];
+    }
+
+    @Transition(from = "OFFLINE", to = "STANDBY")
+    public void becomeLeaderFromOffline(Message msg, NotificationContext context) {
+        logger.info("Transitioning from " + msg.getFromState() + " to " + msg.getToState() + "for "
+                + msg.getPartitionName());
+    }
+
+    @Transition(from = "STANDBY", to = "LEADER")
+    public void becomeLeaderFromStandby(Message msg, NotificationContext context) {
+        logger.info("Transitioning from " + msg.getFromState() + " to " + msg.getToState() + "for "
+                + msg.getPartitionName());
+    }
+
+    @Transition(from = "LEADER", to = "STANDBY")
+    public void becomeStandbyFromLeader(Message msg, NotificationContext context) {
+        logger.info("Transitioning from " + msg.getFromState() + " to " + msg.getToState() + "for "
+                + msg.getPartitionName());
+    }
+
+    @Transition(from = "STANDBY", to = "OFFLINE")
+    public void becomeOfflineFromStandby(Message msg, NotificationContext context) {
+        logger.info("Transitioning from " + msg.getFromState() + " to " + msg.getToState() + "for "
+                + msg.getPartitionName());
+    }
+
+    @Transition(from = "OFFLINE", to = "DROPPED")
+    public void dropPartition(Message msg, NotificationContext context) {
+        logger.info("Dropping partition" + msg.getPartitionName());
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/subprojects/s4-comm/src/main/java/org/apache/s4/comm/helix/TaskStateModelFactory.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/helix/TaskStateModelFactory.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/helix/TaskStateModelFactory.java
new file mode 100644
index 0000000..3b885a8
--- /dev/null
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/helix/TaskStateModelFactory.java
@@ -0,0 +1,12 @@
+package org.apache.s4.comm.helix;
+
+import org.apache.helix.participant.statemachine.StateModelFactory;
+
+public class TaskStateModelFactory extends StateModelFactory<S4StateModel> {
+
+    @Override
+    public S4StateModel createNewStateModel(String partitionName) {
+        return new S4StateModel(partitionName);
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/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 fd0ad2b..d31f04c 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
@@ -277,6 +277,11 @@ public class TCPEmitter implements Emitter, ClusterChangeListener {
     public int getPartitionCount() {
         return topology.getPhysicalCluster().getPartitionCount();
     }
+    
+    @Override
+    public int getPartitionCount(String streamName) {
+        return topology.getPhysicalCluster().getPartitionCount(streamName);
+    }
 
     class ExceptionHandler extends SimpleChannelUpstreamHandler {
         @Override

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/AssignmentFromHelix.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/AssignmentFromHelix.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/AssignmentFromHelix.java
new file mode 100644
index 0000000..243faac
--- /dev/null
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/AssignmentFromHelix.java
@@ -0,0 +1,143 @@
+package org.apache.s4.comm.topology;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.Condition;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+import org.apache.helix.BaseDataAccessor;
+import org.apache.helix.HelixAdmin;
+import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.HelixManager;
+import org.apache.helix.HelixManagerFactory;
+import org.apache.helix.InstanceType;
+import org.apache.helix.PropertyKey.Builder;
+import org.apache.helix.ZNRecord;
+import org.apache.helix.manager.zk.ZKHelixAdmin;
+import org.apache.helix.manager.zk.ZKHelixDataAccessor;
+import org.apache.helix.manager.zk.ZNRecordSerializer;
+import org.apache.helix.manager.zk.ZkBaseDataAccessor;
+import org.apache.helix.manager.zk.ZkClient;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.helix.participant.statemachine.StateModel;
+import org.apache.helix.participant.statemachine.StateModelFactory;
+import org.apache.s4.comm.helix.TaskStateModelFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+import com.google.inject.name.Named;
+
+@Singleton
+public class AssignmentFromHelix implements Assignment {
+    private static final Logger logger = LoggerFactory.getLogger(AssignmentFromHelix.class);
+
+    private String clusterName;
+    private final String zookeeperAddress;
+    private String machineId;
+    private HelixManager zkHelixManager;
+
+    private HelixDataAccessor helixDataAccessor;
+    AtomicReference<ClusterNode> clusterNodeRef;
+    private final Lock lock;
+    private final AtomicBoolean currentlyOwningTask;
+    private final Condition taskAcquired;
+
+    private final StateModelFactory<? extends StateModel> taskStateModelFactory;
+
+    // private final StateModelFactory<? extends StateModel> appStateModelFactory;
+
+    @Inject
+    public AssignmentFromHelix(@Named("s4.cluster.name") String clusterName,
+            @Named("s4.instance.name") String instanceName, @Named("s4.cluster.zk_address") String zookeeperAddress)
+            throws Exception {
+        this.taskStateModelFactory = new TaskStateModelFactory();
+        // this.appStateModelFactory = appStateModelFactory;
+        this.clusterName = clusterName;
+        this.zookeeperAddress = zookeeperAddress;
+        machineId = "localhost";
+        lock = new ReentrantLock();
+        ZkClient zkClient = new ZkClient(zookeeperAddress);
+        zkClient.setZkSerializer(new ZNRecordSerializer());
+        zkClient.waitUntilConnected(60, TimeUnit.SECONDS);
+        BaseDataAccessor<ZNRecord> baseDataAccessor = new ZkBaseDataAccessor<ZNRecord>(zkClient);
+        helixDataAccessor = new ZKHelixDataAccessor(clusterName, baseDataAccessor);
+        clusterNodeRef = new AtomicReference<ClusterNode>();
+        taskAcquired = lock.newCondition();
+        currentlyOwningTask = new AtomicBoolean(true);
+        try {
+            machineId = InetAddress.getLocalHost().getCanonicalHostName();
+        } catch (UnknownHostException e) {
+            logger.warn("Unable to get hostname", e);
+            machineId = "UNKNOWN";
+        }
+        ClusterNode node = new ClusterNode(-1, Integer.parseInt(instanceName.split("_")[1]), machineId, instanceName);
+        clusterNodeRef.set(node);
+        currentlyOwningTask.set(true);
+    }
+
+    @Inject
+    public void init() {
+        // joinCluster();
+    }
+
+    @Override
+    public ClusterNode assignClusterNode() {
+        lock.lock();
+        try {
+            while (!currentlyOwningTask.get()) {
+                taskAcquired.awaitUninterruptibly();
+            }
+        } catch (Exception e) {
+            logger.error("Exception while waiting to join the cluster");
+            return null;
+        } finally {
+            lock.unlock();
+        }
+        return clusterNodeRef.get();
+    }
+
+    public void joinClusterOld() {
+        lock.lock();
+        try {
+            Builder keyBuilder = helixDataAccessor.keyBuilder();
+            do {
+                List<InstanceConfig> instances = helixDataAccessor.getChildValues(keyBuilder.instanceConfigs());
+                List<String> liveInstances = helixDataAccessor.getChildNames(keyBuilder.liveInstances());
+                for (InstanceConfig instanceConfig : instances) {
+                    String instanceName = instanceConfig.getInstanceName();
+                    if (!liveInstances.contains(instanceName)) {
+                        zkHelixManager = HelixManagerFactory.getZKHelixManager(clusterName, instanceName,
+                                InstanceType.PARTICIPANT, zookeeperAddress);
+                        zkHelixManager.getStateMachineEngine().registerStateModelFactory("LeaderStandby",
+                                taskStateModelFactory);
+
+                        zkHelixManager.connect();
+                        ClusterNode node = new ClusterNode(-1, Integer.parseInt(instanceConfig.getPort()), machineId,
+                                instanceName);
+                        clusterNodeRef.set(node);
+                        currentlyOwningTask.set(true);
+                        taskAcquired.signalAll();
+                        break;
+                    }
+                }
+                if (instances.size() == liveInstances.size()) {
+                    System.out.println("No more nodes can join the cluster. Will wait for some node to die.");
+                    Thread.sleep(100000);
+                }
+            } while (!currentlyOwningTask.get());
+            System.out.println("Joined the cluster:" + clusterName + " as " + clusterNodeRef.get().getTaskId());
+        } catch (Exception e) {
+            e.printStackTrace();
+        } finally {
+            lock.unlock();
+        }
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/Cluster.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/Cluster.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/Cluster.java
index 2ff6e97..e3c2eef 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/Cluster.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/Cluster.java
@@ -18,6 +18,8 @@
 
 package org.apache.s4.comm.topology;
 
+import org.apache.helix.model.InstanceConfig;
+
 /**
  * Represents a logical cluster
  * 
@@ -29,4 +31,13 @@ public interface Cluster {
     public void addListener(ClusterChangeListener listener);
 
     public void removeListener(ClusterChangeListener listener);
+    /**
+     * TODO: Use a destination class that provides details on node name, port
+     * @param streamName
+     * @param partitionId
+     * @return
+     */
+    InstanceConfig getDestination(String streamName, int partitionId);
+
+    Integer getPartitionCount(String streamName);
 }

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ClusterFromHelix.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ClusterFromHelix.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ClusterFromHelix.java
new file mode 100644
index 0000000..19e0628
--- /dev/null
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ClusterFromHelix.java
@@ -0,0 +1,190 @@
+/**
+ * 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.comm.topology;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.inject.Inject;
+import com.google.inject.name.Named;
+
+import org.apache.helix.ConfigAccessor;
+import org.apache.helix.ConfigScopeBuilder;
+import org.apache.helix.HelixDataAccessor;
+import org.apache.helix.HelixManager;
+import org.apache.helix.HelixProperty;
+import org.apache.helix.NotificationContext;
+import org.apache.helix.PropertyKey.Builder;
+import org.apache.helix.model.ExternalView;
+import org.apache.helix.model.IdealState;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.helix.spectator.RoutingTableProvider;
+
+/**
+ * Represents a logical cluster definition fetched from Zookeeper. Notifies listeners of runtime changes in the
+ * configuration.
+ * 
+ */
+public class ClusterFromHelix extends RoutingTableProvider implements Cluster {
+
+    private static Logger logger = LoggerFactory.getLogger(ClusterFromHelix.class);
+
+    private final String clusterName;
+    private final AtomicReference<PhysicalCluster> clusterRef;
+    private final List<ClusterChangeListener> listeners;
+    private final Lock lock;
+    private final AtomicReference<Map<String, Integer>> partitionCountMapRef;
+
+    /**
+     * only the local topology
+     */
+    @Inject
+    public ClusterFromHelix(@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;
+        Map<String, Integer> map = Collections.emptyMap();
+        partitionCountMapRef = new AtomicReference<Map<String, Integer>>(map);
+        this.clusterRef = new AtomicReference<PhysicalCluster>();
+        this.listeners = new ArrayList<ClusterChangeListener>();
+        lock = new ReentrantLock();
+
+    }
+
+    /**
+     * any topology
+     */
+    public ClusterFromHelix(String clusterName, ZkClient zkClient, String machineId) {
+        this.clusterName = clusterName;
+        Map<String, Integer> map = Collections.emptyMap();
+        partitionCountMapRef = new AtomicReference<Map<String, Integer>>(map);
+        this.clusterRef = new AtomicReference<PhysicalCluster>();
+        this.listeners = new ArrayList<ClusterChangeListener>();
+        lock = new ReentrantLock();
+
+    }
+
+    @Override
+    public void onExternalViewChange(List<ExternalView> externalViewList, NotificationContext changeContext) {
+        lock.lock();
+        try {
+            logger.info("Start:Processing change in cluster topology");
+            super.onExternalViewChange(externalViewList, changeContext);
+            HelixManager manager = changeContext.getManager();
+            HelixDataAccessor helixDataAccessor = manager.getHelixDataAccessor();
+            ConfigAccessor configAccessor = manager.getConfigAccessor();
+            ConfigScopeBuilder builder = new ConfigScopeBuilder();
+            Builder keyBuilder = helixDataAccessor.keyBuilder();
+            List<String> resources = helixDataAccessor.getChildNames(keyBuilder.idealStates());
+            Map<String, Integer> map = new HashMap<String, Integer>();
+            for (String resource : resources) {
+                String resourceType = configAccessor.get(builder.forCluster(clusterName).forResource(resource).build(),
+                        "type");
+                if ("Task".equalsIgnoreCase(resourceType)) {
+                    String streamName = configAccessor.get(builder.forCluster(clusterName).forResource(resource)
+                            .build(), "streamName");
+                    IdealState idealstate = helixDataAccessor.getProperty(keyBuilder.idealStates(resource));
+                    map.put(streamName, idealstate.getNumPartitions());
+                }
+            }
+            partitionCountMapRef.set(map);
+            for (ClusterChangeListener listener : listeners) {
+                listener.onChange();
+            }
+            logger.info("End:Processing change in cluster topology");
+
+        } catch (Exception e) {
+            logger.error("", e);
+        } finally {
+            lock.unlock();
+        }
+    }
+
+    @Override
+    public PhysicalCluster getPhysicalCluster() {
+        return clusterRef.get();
+    }
+
+    @Override
+    public void addListener(ClusterChangeListener listener) {
+        logger.info("Adding topology change listener:" + listener);
+        listeners.add(listener);
+    }
+
+    @Override
+    public void removeListener(ClusterChangeListener listener) {
+        logger.info("Removing topology change listener:" + listener);
+        listeners.remove(listener);
+    }
+
+    @Override
+    public int hashCode() {
+        final int prime = 31;
+        int result = 1;
+        result = prime * result + ((clusterName == null) ? 0 : clusterName.hashCode());
+        return result;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj)
+            return true;
+        if (obj == null)
+            return false;
+        if (getClass() != obj.getClass())
+            return false;
+        ClusterFromHelix other = (ClusterFromHelix) obj;
+        if (clusterName == null) {
+            if (other.clusterName != null)
+                return false;
+        } else if (!clusterName.equals(other.clusterName))
+            return false;
+        return true;
+    }
+
+    @Override
+    public InstanceConfig getDestination(String streamName, int partitionId) {
+        List<InstanceConfig> instances = getInstances(streamName, streamName + "_" + partitionId, "LEADER");
+        if (instances.size() == 1) {
+            return instances.get(0);
+        } else {
+            return null;
+        }
+    }
+
+    @Override
+    public Integer getPartitionCount(String streamName) {
+        Integer numPartitions = partitionCountMapRef.get().get(streamName);
+        if (numPartitions == null) {
+            return -1;
+        }
+        return numPartitions;
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/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 45e0ae4..f2b60d5 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
@@ -30,6 +30,7 @@ import java.util.concurrent.locks.ReentrantLock;
 import org.I0Itec.zkclient.IZkChildListener;
 import org.I0Itec.zkclient.IZkDataListener;
 import org.I0Itec.zkclient.IZkStateListener;
+import org.apache.helix.model.InstanceConfig;
 import org.apache.zookeeper.Watcher.Event.KeeperState;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -208,5 +209,14 @@ public class ClusterFromZK implements Cluster, IZkChildListener, IZkDataListener
         doProcess();
 
     }
+    @Override
+    public InstanceConfig getDestination(String streamName, int partitionId) {
+        return null;
+    }
+
+    @Override
+    public Integer getPartitionCount(String streamName) {
+        return null;
+    }
 
 }

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ClustersFromHelix.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ClustersFromHelix.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ClustersFromHelix.java
new file mode 100644
index 0000000..4e5351b
--- /dev/null
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ClustersFromHelix.java
@@ -0,0 +1,64 @@
+/**
+ * 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.comm.topology;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
+import org.I0Itec.zkclient.IZkStateListener;
+import org.I0Itec.zkclient.serialize.ZkSerializer;
+import org.apache.zookeeper.Watcher.Event.KeeperState;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.inject.Inject;
+import com.google.inject.name.Named;
+
+/**
+ * Monitors all clusters
+ * 
+ */
+public class ClustersFromHelix implements Clusters {
+    private static final Logger logger = LoggerFactory.getLogger(ClustersFromHelix.class);
+    private String machineId;
+    private Map<String, ClusterFromHelix> clusters = new HashMap<String, ClusterFromHelix>();
+    private int connectionTimeout;
+    private String clusterName;
+
+    @Inject
+    public ClustersFromHelix(@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;
+
+    }
+
+    public Cluster getCluster(String clusterName) {
+        return clusters.get(clusterName);
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/PhysicalCluster.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/PhysicalCluster.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/PhysicalCluster.java
index 7c590ac..233316d 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/PhysicalCluster.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/PhysicalCluster.java
@@ -97,6 +97,14 @@ public class PhysicalCluster {
     public int getPartitionCount() {
         return numPartitions;
     }
+    
+    /**
+     * 
+     * @return Number of partitions in the cluster per stream
+     */
+    public int getPartitionCount(String streamName) {
+        return numPartitions;
+    }
 
     /**
      * @param node

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/subprojects/s4-comm/src/main/java/org/apache/s4/comm/udp/UDPEmitter.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/udp/UDPEmitter.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/udp/UDPEmitter.java
index d335970..b6450b5 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/udp/UDPEmitter.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/udp/UDPEmitter.java
@@ -109,6 +109,11 @@ public class UDPEmitter implements Emitter, ClusterChangeListener {
     public int getPartitionCount() {
         return topology.getPhysicalCluster().getPartitionCount();
     }
+    
+    @Override
+    public int getPartitionCount(String streamName) {
+        return topology.getPhysicalCluster().getPartitionCount(streamName);
+    }
 
     @Override
     public void onChange() {

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/subprojects/s4-core/src/main/java/org/apache/s4/core/BaseModule.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/BaseModule.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/BaseModule.java
index 582f5d0..b9679e7 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/core/BaseModule.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/BaseModule.java
@@ -7,10 +7,15 @@ import org.apache.commons.configuration.ConfigurationConverter;
 import org.apache.commons.configuration.ConfigurationException;
 import org.apache.commons.configuration.PropertiesConfiguration;
 import org.apache.s4.comm.topology.Assignment;
+import org.apache.s4.comm.topology.AssignmentFromHelix;
 import org.apache.s4.comm.topology.AssignmentFromZK;
+import org.apache.s4.comm.topology.Cluster;
+import org.apache.s4.comm.topology.ClusterFromHelix;
 import org.apache.s4.comm.topology.ZkClient;
 import org.apache.s4.comm.util.ArchiveFetcher;
 import org.apache.s4.comm.util.RemoteFileFetcher;
+import org.apache.s4.deploy.DeploymentManager;
+import org.apache.s4.deploy.HelixBasedDeploymentManager;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -26,11 +31,15 @@ public class BaseModule extends AbstractModule {
     private PropertiesConfiguration config;
     InputStream baseConfigInputStream;
     String clusterName;
+    private String instanceName;
+    boolean useHelix = true;
 
-    public BaseModule(InputStream baseConfigInputStream, String clusterName) {
+    public BaseModule(InputStream baseConfigInputStream, String clusterName,
+            String instanceName) {
         super();
         this.baseConfigInputStream = baseConfigInputStream;
         this.clusterName = clusterName;
+        this.instanceName = instanceName;
     }
 
     @Override
@@ -38,17 +47,33 @@ public class BaseModule extends AbstractModule {
         if (config == null) {
             loadProperties(binder());
         }
+        if (useHelix) {
+            bind(Assignment.class).to(AssignmentFromHelix.class)
+                    .asEagerSingleton();
+            bind(Cluster.class).to(ClusterFromHelix.class);
+            bind(DeploymentManager.class).to(HelixBasedDeploymentManager.class).in(Scopes.SINGLETON);
+
+            bind(ArchiveFetcher.class).to(RemoteFileFetcher.class);
+            bind(Bootstrap.class).to(S4HelixBootstrap.class);
+
+            // share the Zookeeper connection
+            bind(ZkClient.class).toProvider(ZkClientProvider.class).in(
+                    Scopes.SINGLETON);
+            return;
+        }
         // a node holds a single partition assignment
-        // ==> Assignment is a singleton so it shared between base, comm and app layers.
+        // ==> Assignment is a singleton so it shared between base, comm and app
+        // layers.
         // it is eager so that the node is able to join a cluster immediately
         bind(Assignment.class).to(AssignmentFromZK.class).asEagerSingleton();
         // bind(Cluster.class).to(ClusterFromZK.class);
 
         bind(ArchiveFetcher.class).to(RemoteFileFetcher.class);
-        bind(S4Bootstrap.class);
+        bind(Bootstrap.class).to(S4Bootstrap.class);
 
         // share the Zookeeper connection
-        bind(ZkClient.class).toProvider(ZkClientProvider.class).in(Scopes.SINGLETON);
+        bind(ZkClient.class).toProvider(ZkClientProvider.class).in(
+                Scopes.SINGLETON);
 
     }
 
@@ -61,7 +86,8 @@ public class BaseModule extends AbstractModule {
             // TODO - validate properties.
 
             /* Make all properties injectable. Do we need this? */
-            Names.bindProperties(binder, ConfigurationConverter.getProperties(config));
+            Names.bindProperties(binder,
+                    ConfigurationConverter.getProperties(config));
 
             if (clusterName != null) {
                 if (config.containsKey("s4.cluster.name")) {
@@ -76,6 +102,20 @@ public class BaseModule extends AbstractModule {
                     });
                 }
             }
+            if (instanceName != null) {
+                if (config.containsKey("s4.instance.name")) {
+                    logger.warn(
+                            "instanceName [{}] passed as a parameter will not be used because an existing s4.instance.name parameter of value [{}] was found in the configuration file and will be used",
+                            instanceName,
+                            config.getProperty("s4.instance.name"));
+                } else {
+                    Names.bindProperties(binder, new HashMap<String, String>() {
+                        {
+                            put("s4.instance.name", instanceName);
+                        }
+                    });
+                }
+            }
 
         } catch (ConfigurationException e) {
             binder.addError(e);

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/subprojects/s4-core/src/main/java/org/apache/s4/core/Bootstrap.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/Bootstrap.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/Bootstrap.java
new file mode 100644
index 0000000..3bc818b
--- /dev/null
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/Bootstrap.java
@@ -0,0 +1,7 @@
+package org.apache.s4.core;
+
+import com.google.inject.Injector;
+
+public interface Bootstrap {
+	void start(Injector parentInjector) throws Exception;
+}

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/subprojects/s4-core/src/main/java/org/apache/s4/core/DefaultCoreModule.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/DefaultCoreModule.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/DefaultCoreModule.java
index c4cafee..30e44b2 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/core/DefaultCoreModule.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/DefaultCoreModule.java
@@ -25,6 +25,7 @@ import java.io.InputStream;
 import org.apache.commons.configuration.ConfigurationConverter;
 import org.apache.commons.configuration.ConfigurationException;
 import org.apache.commons.configuration.PropertiesConfiguration;
+import org.apache.helix.participant.statemachine.StateModelFactory;
 import org.apache.s4.base.Hasher;
 import org.apache.s4.base.util.S4RLoaderFactory;
 import org.apache.s4.comm.DefaultHasher;
@@ -38,8 +39,10 @@ import org.apache.s4.core.staging.RemoteSendersExecutorServiceFactory;
 import org.apache.s4.core.staging.SenderExecutorServiceFactory;
 import org.apache.s4.core.staging.StreamExecutorServiceFactory;
 import org.apache.s4.core.staging.ThrottlingSenderExecutorServiceFactory;
+import org.apache.s4.deploy.AppStateModelFactory;
 import org.apache.s4.deploy.DeploymentManager;
 import org.apache.s4.deploy.DistributedDeploymentManager;
+import org.apache.s4.deploy.HelixBasedDeploymentManager;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -84,7 +87,10 @@ public class DefaultCoreModule extends AbstractModule {
         /* The hashing function to map keys top partitions. */
         bind(Hasher.class).to(DefaultHasher.class);
 
-        bind(DeploymentManager.class).to(DistributedDeploymentManager.class).in(Scopes.SINGLETON);
+        bind(StateModelFactory.class).annotatedWith(Names.named("s4.app.statemodelfactory")).to(
+                AppStateModelFactory.class);
+        
+        bind(DeploymentManager.class).to(HelixBasedDeploymentManager.class).in(Scopes.SINGLETON);
 
         bind(S4RLoaderFactory.class);
 

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/subprojects/s4-core/src/main/java/org/apache/s4/core/HelixBasedCoreModule.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/HelixBasedCoreModule.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/HelixBasedCoreModule.java
new file mode 100644
index 0000000..0ce80ed
--- /dev/null
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/HelixBasedCoreModule.java
@@ -0,0 +1,134 @@
+/**
+ * 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.core;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+
+import org.apache.commons.configuration.ConfigurationConverter;
+import org.apache.commons.configuration.ConfigurationException;
+import org.apache.commons.configuration.PropertiesConfiguration;
+import org.apache.s4.base.Hasher;
+import org.apache.s4.base.util.S4RLoaderFactory;
+import org.apache.s4.comm.DefaultHasher;
+import org.apache.s4.comm.topology.RemoteStreams;
+import org.apache.s4.comm.topology.ZkRemoteStreams;
+import org.apache.s4.core.ft.CheckpointingFramework;
+import org.apache.s4.core.ft.NoOpCheckpointingFramework;
+import org.apache.s4.core.staging.BlockingRemoteSendersExecutorServiceFactory;
+import org.apache.s4.core.staging.BlockingStreamExecutorServiceFactory;
+import org.apache.s4.core.staging.RemoteSendersExecutorServiceFactory;
+import org.apache.s4.core.staging.SenderExecutorServiceFactory;
+import org.apache.s4.core.staging.StreamExecutorServiceFactory;
+import org.apache.s4.core.staging.ThrottlingSenderExecutorServiceFactory;
+import org.apache.s4.deploy.DeploymentManager;
+import org.apache.s4.deploy.DistributedDeploymentManager;
+import org.apache.s4.deploy.HelixBasedDeploymentManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.io.Files;
+import com.google.inject.AbstractModule;
+import com.google.inject.Binder;
+import com.google.inject.Provides;
+import com.google.inject.Scopes;
+import com.google.inject.name.Named;
+import com.google.inject.name.Names;
+
+/**
+ * Temporary module allowing assignment from ZK, communication through Netty, and distributed deployment management,
+ * until we have a better way to customize node configuration
+ * 
+ */
+public class HelixBasedCoreModule extends AbstractModule {
+
+    private static Logger logger = LoggerFactory.getLogger(HelixBasedCoreModule.class);
+
+    InputStream coreConfigFileInputStream;
+    private PropertiesConfiguration config;
+
+    String clusterName = null;
+
+    public HelixBasedCoreModule(InputStream coreConfigFileInputStream) {
+        this.coreConfigFileInputStream = coreConfigFileInputStream;
+    }
+
+    @Override
+    protected void configure() {
+        if (config == null) {
+            loadProperties(binder());
+        }
+        if (coreConfigFileInputStream != null) {
+            try {
+                coreConfigFileInputStream.close();
+            } catch (IOException ignored) {
+            }
+        }
+
+        /* The hashing function to map keys top partitions. */
+        bind(Hasher.class).to(DefaultHasher.class);
+
+        bind(DeploymentManager.class).to(HelixBasedDeploymentManager.class).in(Scopes.SINGLETON);
+
+        bind(S4RLoaderFactory.class);
+
+        // For enabling checkpointing, one needs to use a custom module, such as
+        // org.apache.s4.core.ft.FileSytemBasedCheckpointingModule
+        bind(CheckpointingFramework.class).to(NoOpCheckpointingFramework.class);
+
+        // shed load in local sender only by default
+        bind(SenderExecutorServiceFactory.class).to(ThrottlingSenderExecutorServiceFactory.class);
+        bind(RemoteSendersExecutorServiceFactory.class).to(BlockingRemoteSendersExecutorServiceFactory.class);
+
+        bind(StreamExecutorServiceFactory.class).to(BlockingStreamExecutorServiceFactory.class);
+
+        bind(RemoteStreams.class).to(ZkRemoteStreams.class).in(Scopes.SINGLETON);
+        bind(RemoteSenders.class).to(DefaultRemoteSenders.class).in(Scopes.SINGLETON);
+
+    }
+
+    @Provides
+    @Named("s4.tmp.dir")
+    public File provideTmpDir() {
+        File tmpS4Dir = Files.createTempDir();
+        tmpS4Dir.deleteOnExit();
+        logger.warn(
+                "s4.tmp.dir not specified, using temporary directory [{}] for unpacking S4R. You may want to specify a parent non-temporary directory.",
+                tmpS4Dir.getAbsolutePath());
+        return tmpS4Dir;
+    }
+
+    private void loadProperties(Binder binder) {
+        try {
+            config = new PropertiesConfiguration();
+            config.load(coreConfigFileInputStream);
+
+            // 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();
+        }
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/subprojects/s4-core/src/main/java/org/apache/s4/core/S4Bootstrap.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/S4Bootstrap.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/S4Bootstrap.java
index 77dbc67..61f6b7f 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/core/S4Bootstrap.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/S4Bootstrap.java
@@ -58,7 +58,7 @@ import com.google.inject.util.Modules.OverriddenModuleBuilder;
  * 
  * 
  */
-public class S4Bootstrap {
+public class S4Bootstrap implements Bootstrap{
     private static Logger logger = LoggerFactory.getLogger(S4Bootstrap.class);
 
     private final ZkClient zkClient;

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/subprojects/s4-core/src/main/java/org/apache/s4/core/S4HelixBootstrap.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/S4HelixBootstrap.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/S4HelixBootstrap.java
new file mode 100644
index 0000000..d20c962
--- /dev/null
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/S4HelixBootstrap.java
@@ -0,0 +1,139 @@
+package org.apache.s4.core;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStream;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.I0Itec.zkclient.IZkDataListener;
+import org.I0Itec.zkclient.serialize.ZkSerializer;
+import org.apache.helix.HelixManager;
+import org.apache.helix.HelixManagerFactory;
+import org.apache.helix.InstanceType;
+import org.apache.helix.spectator.RoutingTableProvider;
+import org.apache.s4.base.util.ModulesLoader;
+import org.apache.s4.comm.DefaultCommModule;
+import org.apache.s4.comm.ModulesLoaderFactory;
+import org.apache.s4.comm.helix.TaskStateModelFactory;
+import org.apache.s4.comm.topology.Cluster;
+import org.apache.s4.comm.topology.ZNRecord;
+import org.apache.s4.comm.topology.ZNRecordSerializer;
+import org.apache.s4.comm.topology.ZkClient;
+import org.apache.s4.comm.util.ArchiveFetchException;
+import org.apache.s4.comm.util.ArchiveFetcher;
+import org.apache.s4.core.util.AppConfig;
+import org.apache.s4.core.util.ParametersInjectionModule;
+import org.apache.s4.deploy.AppStateModelFactory;
+import org.apache.zookeeper.CreateMode;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.base.Strings;
+import com.google.common.io.ByteStreams;
+import com.google.common.io.Files;
+import com.google.common.io.Resources;
+import com.google.inject.AbstractModule;
+import com.google.inject.Inject;
+import com.google.inject.Injector;
+import com.google.inject.Module;
+import com.google.inject.name.Named;
+import com.google.inject.util.Modules;
+import com.google.inject.util.Modules.OverriddenModuleBuilder;
+
+/**
+ * This is the bootstrap for S4 nodes.
+ * <p>
+ * Its roles are to:
+ * <ul>
+ * <li>register within the S4 cluster (and acquire a partition).
+ * <li>look for application deployed on the S4 cluster
+ * </ul>
+ * <p>
+ * When an application is available, custom modules are fetched if necessary and
+ * a full-featured S4 node is started. The application code is then downloaded
+ * and the app started.
+ * <p>
+ * For testing purposes, it is also possible to directly start an application
+ * without fetching remote code, provided the application classes are available
+ * in the classpath.
+ * 
+ * 
+ * 
+ */
+public class S4HelixBootstrap implements Bootstrap{
+	private static Logger logger = LoggerFactory
+			.getLogger(S4HelixBootstrap.class);
+
+	private final AtomicBoolean deployed = new AtomicBoolean(false);
+
+	private final ArchiveFetcher fetcher;
+
+	private Injector parentInjector;
+
+	CountDownLatch signalOneAppLoaded = new CountDownLatch(1);
+
+	private String clusterName;
+
+	private String instanceName;
+
+	private String zookeeperAddress;
+    @Inject
+    private TaskStateModelFactory taskStateModelFactory;
+    
+    @Inject
+    private AppStateModelFactory appStateModelFactory;
+    
+    @Inject
+    private Cluster cluster;
+
+	@Inject
+	public S4HelixBootstrap(@Named("s4.cluster.name") String clusterName,
+            @Named("s4.instance.name") String instanceName,
+			@Named("s4.cluster.zk_address") String zookeeperAddress,
+			@Named("s4.cluster.zk_session_timeout") int sessionTimeout,
+			@Named("s4.cluster.zk_connection_timeout") int connectionTimeout,
+			ArchiveFetcher fetcher) {
+        this.clusterName = clusterName;
+        this.instanceName = instanceName;
+        this.zookeeperAddress = zookeeperAddress;
+		this.fetcher = fetcher;
+        registerWithHelix();
+	}
+
+	public void start(Injector parentInjector) throws InterruptedException,
+			ArchiveFetchException {
+		this.parentInjector = parentInjector;
+		if (!deployed.get()) {
+
+		}
+		signalOneAppLoaded.await();
+	}
+	
+	private void registerWithHelix()
+    {
+      HelixManager helixManager;
+      try
+      {
+        helixManager = HelixManagerFactory.getZKHelixManager(clusterName,
+            instanceName, InstanceType.PARTICIPANT, zookeeperAddress);
+        helixManager.getStateMachineEngine().registerStateModelFactory(
+          "LeaderStandby", taskStateModelFactory);
+        helixManager.getStateMachineEngine().registerStateModelFactory(
+          "OnlineOffline", appStateModelFactory);
+        helixManager.connect();  
+        helixManager.addExternalViewChangeListener((RoutingTableProvider)cluster);
+      } catch (Exception e)
+      {
+        // TODO Auto-generated catch block
+        e.printStackTrace();
+      }
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/subprojects/s4-core/src/main/java/org/apache/s4/core/S4Node.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/S4Node.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/S4Node.java
index 907fcb9..db84c35 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/core/S4Node.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/S4Node.java
@@ -50,7 +50,7 @@ public class S4Node {
         });
 
         Injector injector = Guice.createInjector(new Module[] { new BaseModule(Resources.getResource(
-                "default.s4.base.properties").openStream(), nodeArgs.clusterName) });
+                "default.s4.base.properties").openStream(), nodeArgs.clusterName, nodeArgs.instanceName) });
         S4Bootstrap bootstrap = injector.getInstance(S4Bootstrap.class);
         try {
             bootstrap.start(injector);
@@ -74,6 +74,8 @@ public class S4Node {
 
         @Parameter(names = "-zk", description = "Zookeeper connection string", required = false)
         String zkConnectionString;
-
+        
+        @Parameter(names = { "-id", "-nodeId" }, description = "Node/Instance id that uniquely identifies a node", required = false)
+        String instanceName = null;
     }
 }

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/subprojects/s4-core/src/main/java/org/apache/s4/deploy/AppStateModel.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/deploy/AppStateModel.java b/subprojects/s4-core/src/main/java/org/apache/s4/deploy/AppStateModel.java
new file mode 100644
index 0000000..5f21f51
--- /dev/null
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/deploy/AppStateModel.java
@@ -0,0 +1,73 @@
+package org.apache.s4.deploy;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.helix.ConfigAccessor;
+import org.apache.helix.ConfigScope;
+import org.apache.helix.ConfigScopeBuilder;
+import org.apache.helix.HelixManager;
+import org.apache.helix.NotificationContext;
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.statemachine.StateModel;
+import org.apache.helix.participant.statemachine.StateModelInfo;
+import org.apache.helix.participant.statemachine.Transition;
+import org.apache.s4.core.util.AppConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+@StateModelInfo(states = { "ONLINE,OFFLINE" }, initialState = "OFFLINE")
+public class AppStateModel extends StateModel {
+	private static Logger logger = LoggerFactory.getLogger(AppStateModel.class);
+	private final String appName;
+	private DeploymentManager deploymentManager;
+
+	public AppStateModel(DeploymentManager deploymentManager, String appName) {
+		this.deploymentManager = deploymentManager;
+		this.appName = appName;
+	}
+
+	@Transition(from = "OFFLINE", to = "ONLINE")
+	public void deploy(Message message, NotificationContext context)
+			throws Exception {
+		logger.info("Deploying app:"+ appName);
+		HelixManager manager = context.getManager();
+		ConfigAccessor configAccessor = manager.getConfigAccessor();
+		AppConfig config = createAppConfig(manager, configAccessor);
+		deploymentManager.deploy(config);
+		logger.info("Deployed app:"+ appName);
+
+	}
+
+	private AppConfig createAppConfig(HelixManager manager,
+			ConfigAccessor configAccessor) {
+		ConfigScopeBuilder builder = new ConfigScopeBuilder();
+		ConfigScope scope = builder.forCluster(manager.getClusterName())
+				.forResource(appName).build();
+		String appURI = configAccessor.get(scope,
+				DeploymentManager.S4R_URI);
+		String clusterName = manager.getClusterName();
+		String appClassName = null;
+		List<String> customModulesNames = new ArrayList<String>();
+		List<String> customModulesURIs = new ArrayList<String>();
+		Map<String, String> namedParameters = new HashMap<String, String>();
+		AppConfig config = new AppConfig(clusterName, appClassName, appURI,
+				customModulesNames, customModulesURIs, namedParameters);
+		return config;
+	}
+
+	@Transition(from = "OFFLINE", to = "ONLINE")
+	public void undeploy(Message message, NotificationContext context)
+			throws Exception {
+		logger.info("Undeploying app:"+ appName);
+		HelixManager manager = context.getManager();
+		ConfigAccessor configAccessor = manager.getConfigAccessor();
+		AppConfig config = createAppConfig(manager, configAccessor);
+		deploymentManager.undeploy(config);
+		logger.info("Undeploying app:"+ appName);
+
+	}
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/subprojects/s4-core/src/main/java/org/apache/s4/deploy/AppStateModelFactory.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/deploy/AppStateModelFactory.java b/subprojects/s4-core/src/main/java/org/apache/s4/deploy/AppStateModelFactory.java
new file mode 100644
index 0000000..4f6dc00
--- /dev/null
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/deploy/AppStateModelFactory.java
@@ -0,0 +1,25 @@
+package org.apache.s4.deploy;
+
+import org.apache.helix.participant.statemachine.StateModelFactory;
+import org.apache.s4.comm.util.ArchiveFetcher;
+import org.apache.s4.core.Server;
+
+import com.google.inject.Inject;
+import com.google.inject.Singleton;
+
+@Singleton
+public class AppStateModelFactory extends StateModelFactory<AppStateModel> {
+    private final DeploymentManager deploymentManager;
+
+    @Inject
+    public AppStateModelFactory(DeploymentManager deploymentManager, ArchiveFetcher fetcher) {
+        this.deploymentManager = deploymentManager;
+
+    }
+
+    @Override
+    public AppStateModel createNewStateModel(String partitionName) {
+        return new AppStateModel(deploymentManager,partitionName);
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/subprojects/s4-core/src/main/java/org/apache/s4/deploy/DeploymentManager.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/deploy/DeploymentManager.java b/subprojects/s4-core/src/main/java/org/apache/s4/deploy/DeploymentManager.java
index e92e97d..1b8994e 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/deploy/DeploymentManager.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/deploy/DeploymentManager.java
@@ -18,13 +18,20 @@
 
 package org.apache.s4.deploy;
 
+import org.apache.s4.core.util.AppConfig;
+
 /**
  * Marker interface for deployment managers. Allows to supply a no-op deployment manager through dependency injection.
  * (TODO that hack should be improved!)
  * 
  */
 public interface DeploymentManager {
-
+    public static final String S4R_URI = "s4r_uri";
+    
     void start();
+    
+    void deploy(AppConfig appConfig) throws DeploymentFailedException;
+    
+    void undeploy(AppConfig appConfig) throws DeploymentFailedException;
 
 }

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/subprojects/s4-core/src/main/java/org/apache/s4/deploy/DeploymentUtils.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/deploy/DeploymentUtils.java b/subprojects/s4-core/src/main/java/org/apache/s4/deploy/DeploymentUtils.java
index cc404af..fc42754 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/deploy/DeploymentUtils.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/deploy/DeploymentUtils.java
@@ -1,14 +1,25 @@
 package org.apache.s4.deploy;
 
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+
 import org.I0Itec.zkclient.exception.ZkNodeExistsException;
 import org.I0Itec.zkclient.serialize.ZkSerializer;
 import org.apache.s4.comm.topology.ZNRecordSerializer;
 import org.apache.s4.comm.topology.ZkClient;
+import org.apache.s4.comm.util.ArchiveFetcher;
+import org.apache.s4.core.App;
+import org.apache.s4.core.Server;
 import org.apache.s4.core.util.AppConfig;
 import org.apache.zookeeper.CreateMode;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import com.google.common.io.ByteStreams;
+import com.google.common.io.Files;
+
 public class DeploymentUtils {
 
     private static Logger logger = LoggerFactory.getLogger(DeploymentUtils.class);
@@ -34,5 +45,59 @@ public class DeploymentUtils {
         }
         zk.close();
     }
+	public static void deploy(Server server, ArchiveFetcher fetcher, String clusterName, AppConfig appConfig) throws DeploymentFailedException {
+        if (appConfig.getAppURI() == null) {
+            if (appConfig.getAppClassName() != null) {
+                try {
+                    App app = (App) DeploymentUtils.class.getClassLoader().loadClass(appConfig.getAppClassName()).newInstance();
+                    server.startApp(app, "appName", clusterName);
+                } catch (Exception e) {
+                    logger.error("Cannot start application: cannot instantiate app class {} due to: {}",
+                            appConfig.getAppClassName(), e.getMessage());
+                    return;
+                }
+            }
+            logger.info("{} value not set for {} : no application code will be downloaded", DeploymentManager.S4R_URI, appConfig.getAppName());
+            return;
+        }
+        try {
+            URI uri = new URI(appConfig.getAppURI());
 
+            // fetch application
+            File localS4RFileCopy;
+            try {
+                localS4RFileCopy = File.createTempFile("tmp", "s4r");
+            } catch (IOException e1) {
+                logger.error(
+                        "Cannot deploy app [{}] because a local copy of the S4R file could not be initialized due to [{}]",
+                        appConfig.getAppName(), e1.getClass().getName() + "->" + e1.getMessage());
+                throw new DeploymentFailedException("Cannot deploy application [" + appConfig.getAppName() + "]", e1);
+            }
+            localS4RFileCopy.deleteOnExit();
+            try {
+                if (ByteStreams.copy(fetcher.fetch(uri), Files.newOutputStreamSupplier(localS4RFileCopy)) == 0) {
+                    throw new DeploymentFailedException("Cannot copy archive from [" + uri.toString() + "] to ["
+                            + localS4RFileCopy.getAbsolutePath() + "] (nothing was copied)");
+                }
+            } catch (Exception e) {
+                throw new DeploymentFailedException("Cannot deploy application [" + appConfig.getAppName()
+                        + "] from URI [" + uri.toString() + "] ", e);
+            }
+            // install locally
+            App loaded = server.loadApp(localS4RFileCopy, appConfig.getAppName());
+            if (loaded != null) {
+                logger.info("Successfully installed application {}", appConfig.getAppName());
+                // TODO sync with other nodes? (e.g. wait for other apps deployed before starting?
+                server.startApp(loaded, appConfig.getAppName(), clusterName);
+            } else {
+                throw new DeploymentFailedException("Cannot deploy application [" + appConfig.getAppName()
+                        + "] from URI [" + uri.toString() + "] : cannot start application");
+            }
+
+        } catch (URISyntaxException e) {
+            logger.error("Cannot deploy app {} : invalid uri for fetching s4r archive {} : {} ", new String[] {
+                    appConfig.getAppName(), appConfig.getAppURI(), e.getMessage() });
+            throw new DeploymentFailedException("Cannot deploy application [" + appConfig.getAppName() + "]", e);
+        }
+	}
 }

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/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 fc05c48..60d66bc 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
@@ -68,7 +68,6 @@ import com.google.inject.name.Named;
  */
 public class DistributedDeploymentManager implements DeploymentManager {
 
-    public static final String S4R_URI = "s4r_uri";
 
     private static Logger logger = LoggerFactory.getLogger(DistributedDeploymentManager.class);
 
@@ -103,63 +102,14 @@ public class DistributedDeploymentManager implements DeploymentManager {
 
     public void deployApplication() throws DeploymentFailedException {
         ZNRecord appData = zkClient.readData(appPath);
-        AppConfig appConfig = new AppConfig(appData);
-        if (appConfig.getAppURI() == null) {
-            if (appConfig.getAppClassName() != null) {
-                try {
-                    App app = (App) getClass().getClassLoader().loadClass(appConfig.getAppClassName()).newInstance();
-                    server.startApp(app, "appName", clusterName);
-                } catch (Exception e) {
-                    logger.error("Cannot start application: cannot instantiate app class {} due to: {}",
-                            appConfig.getAppClassName(), e.getMessage());
-                    return;
-                }
-            }
-            logger.info("{} value not set for {} : no application code will be downloaded", S4R_URI, appPath);
-            return;
-        }
-        try {
-            URI uri = new URI(appConfig.getAppURI());
-
-            // fetch application
-            File localS4RFileCopy;
-            try {
-                localS4RFileCopy = File.createTempFile("tmp", "s4r");
-            } catch (IOException e1) {
-                logger.error(
-                        "Cannot deploy app [{}] because a local copy of the S4R file could not be initialized due to [{}]",
-                        appConfig.getAppName(), e1.getClass().getName() + "->" + e1.getMessage());
-                throw new DeploymentFailedException("Cannot deploy application [" + appConfig.getAppName() + "]", e1);
-            }
-            localS4RFileCopy.deleteOnExit();
-            try {
-                if (ByteStreams.copy(fetcher.fetch(uri), Files.newOutputStreamSupplier(localS4RFileCopy)) == 0) {
-                    throw new DeploymentFailedException("Cannot copy archive from [" + uri.toString() + "] to ["
-                            + localS4RFileCopy.getAbsolutePath() + "] (nothing was copied)");
-                }
-            } catch (Exception e) {
-                throw new DeploymentFailedException("Cannot deploy application [" + appConfig.getAppName()
-                        + "] from URI [" + uri.toString() + "] ", e);
-            }
-            // install locally
-            App loaded = server.loadApp(localS4RFileCopy, appConfig.getAppName());
-            if (loaded != null) {
-                logger.info("Successfully installed application {}", appConfig.getAppName());
-                // TODO sync with other nodes? (e.g. wait for other apps deployed before starting?
-                server.startApp(loaded, appConfig.getAppName(), clusterName);
-            } else {
-                throw new DeploymentFailedException("Cannot deploy application [" + appConfig.getAppName()
-                        + "] from URI [" + uri.toString() + "] : cannot start application");
-            }
-
-        } catch (URISyntaxException e) {
-            logger.error("Cannot deploy app {} : invalid uri for fetching s4r archive {} : {} ", new String[] {
-                    appConfig.getAppName(), appConfig.getAppURI(), e.getMessage() });
-            throw new DeploymentFailedException("Cannot deploy application [" + appConfig.getAppName() + "]", e);
-        }
+		AppConfig appConfig = new AppConfig(appData);
+        deploy(appConfig);
+        DeploymentUtils.deploy(server, fetcher, clusterName, appConfig);
         deployed = true;
     }
 
+
+
     // NOTE: in theory, we could support any protocol by implementing a chained visitor scheme,
     // but that's probably not that useful, and we can simply provide whichever protocol is needed
 
@@ -191,4 +141,16 @@ public class DistributedDeploymentManager implements DeploymentManager {
             }
         }
     }
+
+	@Override
+	public void deploy(AppConfig appConfig) {
+		// TODO Auto-generated method stub
+		
+	}
+
+	@Override
+	public void undeploy(AppConfig appConfig) {
+		// TODO Auto-generated method stub
+		
+	}
 }

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/subprojects/s4-core/src/main/java/org/apache/s4/deploy/HelixBasedDeploymentManager.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/deploy/HelixBasedDeploymentManager.java b/subprojects/s4-core/src/main/java/org/apache/s4/deploy/HelixBasedDeploymentManager.java
new file mode 100644
index 0000000..caec699
--- /dev/null
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/deploy/HelixBasedDeploymentManager.java
@@ -0,0 +1,42 @@
+package org.apache.s4.deploy;
+
+import org.apache.s4.comm.util.ArchiveFetcher;
+import org.apache.s4.core.Server;
+import org.apache.s4.core.util.AppConfig;
+
+import com.google.inject.Inject;
+import com.google.inject.name.Named;
+
+public class HelixBasedDeploymentManager implements DeploymentManager {
+    private final Server server;
+    boolean deployed = false;
+    private final String clusterName;
+	private ArchiveFetcher fetcher;
+
+    @Inject
+    public HelixBasedDeploymentManager(@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, ArchiveFetcher fetcher) {
+        this.clusterName = clusterName;
+        this.server = server;
+		this.fetcher = fetcher;
+
+    }
+
+    @Override
+    public void start() {
+        
+    }
+
+	@Override
+	public void deploy(AppConfig appConfig) throws DeploymentFailedException {
+		DeploymentUtils.deploy(server, fetcher, clusterName, appConfig);
+	}
+
+	@Override
+	public void undeploy(AppConfig appConfig) {
+		
+	}
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/subprojects/s4-core/src/main/java/org/apache/s4/deploy/NoOpDeploymentManager.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/deploy/NoOpDeploymentManager.java b/subprojects/s4-core/src/main/java/org/apache/s4/deploy/NoOpDeploymentManager.java
index 27777d8..4aea928 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/deploy/NoOpDeploymentManager.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/deploy/NoOpDeploymentManager.java
@@ -18,6 +18,8 @@
 
 package org.apache.s4.deploy;
 
+import org.apache.s4.core.util.AppConfig;
+
 /**
  * Does not handle any deployment (hence does not require any cluster configuration settings)
  * 
@@ -28,4 +30,16 @@ public class NoOpDeploymentManager implements DeploymentManager {
     public void start() {
         // does nothing
     }
+
+	@Override
+	public void deploy(AppConfig appConfig) {
+		// TODO Auto-generated method stub
+		
+	}
+
+	@Override
+	public void undeploy(AppConfig appConfig) {
+		// TODO Auto-generated method stub
+		
+	}
 }

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/subprojects/s4-core/src/test/java/org/apache/s4/core/moduleloader/ModuleLoaderTestUtils.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/test/java/org/apache/s4/core/moduleloader/ModuleLoaderTestUtils.java b/subprojects/s4-core/src/test/java/org/apache/s4/core/moduleloader/ModuleLoaderTestUtils.java
index e5d0d01..1ca862c 100644
--- a/subprojects/s4-core/src/test/java/org/apache/s4/core/moduleloader/ModuleLoaderTestUtils.java
+++ b/subprojects/s4-core/src/test/java/org/apache/s4/core/moduleloader/ModuleLoaderTestUtils.java
@@ -68,7 +68,7 @@ public class ModuleLoaderTestUtils {
         }
 
         Injector injector = Guice.createInjector(new BaseModule(Resources.getResource("default.s4.base.properties")
-                .openStream(), "cluster1"), new DefaultCommModule(Resources.getResource("default.s4.comm.properties")
+                .openStream(), "cluster1",null), new DefaultCommModule(Resources.getResource("default.s4.comm.properties")
                 .openStream()));
 
         Emitter emitter = injector.getInstance(TCPEmitter.class);

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/CoreTestUtils.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/CoreTestUtils.java b/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/CoreTestUtils.java
index 9b24f3c..965a621 100644
--- a/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/CoreTestUtils.java
+++ b/subprojects/s4-core/src/test/java/org/apache/s4/fixtures/CoreTestUtils.java
@@ -116,7 +116,7 @@ public class CoreTestUtils extends CommTestUtils {
 
     public static Injector createInjectorWithNonFailFastZKClients() throws IOException {
         return Guice.createInjector(Modules.override(
-                new BaseModule(Resources.getResource("default.s4.base.properties").openStream(), "cluster1"),
+                new BaseModule(Resources.getResource("default.s4.base.properties").openStream(), "cluster1", null),
                 new DefaultCommModule(Resources.getResource("default.s4.comm.properties").openStream()),
                 new DefaultCoreModule(Resources.getResource("default.s4.core.properties").openStream())).with(
                 new NonFailFastZookeeperClientsModule()));

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/subprojects/s4-core/src/test/java/org/apache/s4/wordcount/WordCountTest.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-core/src/test/java/org/apache/s4/wordcount/WordCountTest.java b/subprojects/s4-core/src/test/java/org/apache/s4/wordcount/WordCountTest.java
index 5b4f596..c25d756 100644
--- a/subprojects/s4-core/src/test/java/org/apache/s4/wordcount/WordCountTest.java
+++ b/subprojects/s4-core/src/test/java/org/apache/s4/wordcount/WordCountTest.java
@@ -59,7 +59,7 @@ public class WordCountTest extends ZkBasedTest {
 
     public void createEmitter() throws IOException {
         injector = Guice.createInjector(new BaseModule(
-                Resources.getResource("default.s4.base.properties").openStream(), "cluster1"), new DefaultCommModule(
+                Resources.getResource("default.s4.base.properties").openStream(), "cluster1", null), new DefaultCommModule(
                 Resources.getResource("default.s4.comm.properties").openStream()), new DefaultCoreModule(Resources
                 .getResource("default.s4.core.properties").openStream()));
 

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/subprojects/s4-tools/src/main/java/org/apache/s4/tools/Tools.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-tools/src/main/java/org/apache/s4/tools/Tools.java b/subprojects/s4-tools/src/main/java/org/apache/s4/tools/Tools.java
index 832cfcc..d25f7a7 100644
--- a/subprojects/s4-tools/src/main/java/org/apache/s4/tools/Tools.java
+++ b/subprojects/s4-tools/src/main/java/org/apache/s4/tools/Tools.java
@@ -26,6 +26,13 @@ import java.util.List;
 import org.apache.log4j.BasicConfigurator;
 import org.apache.log4j.Level;
 import org.apache.s4.core.S4Node;
+import org.apache.s4.tools.helix.AddNodes;
+import org.apache.s4.tools.helix.CreateCluster;
+import org.apache.s4.tools.helix.CreateTask;
+import org.apache.s4.tools.helix.DeployApp;
+import org.apache.s4.tools.helix.GenericEventAdapter;
+import org.apache.s4.tools.helix.RebalanceTask;
+import org.apache.s4.tools.helix.S4Status;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -39,9 +46,14 @@ public class Tools {
     static Logger logger = LoggerFactory.getLogger(Tools.class);
 
     enum Task {
-        deploy(Deploy.class), node(S4Node.class), zkServer(ZKServer.class), newCluster(DefineCluster.class), adapter(
-                null), newApp(CreateApp.class), s4r(Package.class), status(Status.class);
-
+        //deploy(Deploy.class), node(S4Node.class), zkServer(ZKServer.class), newCluster(DefineCluster.class), adapter(
+          //      null), newApp(CreateApp.class), s4r(Package.class), status(Status.class);
+
+        deployApp(DeployApp.class), node(S4Node.class), zkServer(ZKServer.class), newCluster(CreateCluster.class), genericAdapter(
+                GenericEventAdapter.class), newApp(CreateApp.class), s4r(Package.class), status(S4Status.class),
+                addNodes(AddNodes.class),createTask(
+                        CreateTask.class), rebalanceTask(RebalanceTask.class);
+        
         Class<?> target;
 
         Task(Class<?> target) {

http://git-wip-us.apache.org/repos/asf/incubator-s4/blob/55f8c215/subprojects/s4-tools/src/main/java/org/apache/s4/tools/helix/AddNodes.java
----------------------------------------------------------------------
diff --git a/subprojects/s4-tools/src/main/java/org/apache/s4/tools/helix/AddNodes.java b/subprojects/s4-tools/src/main/java/org/apache/s4/tools/helix/AddNodes.java
new file mode 100644
index 0000000..136a4aa
--- /dev/null
+++ b/subprojects/s4-tools/src/main/java/org/apache/s4/tools/helix/AddNodes.java
@@ -0,0 +1,90 @@
+/**
+ * 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.tools.helix;
+
+import org.apache.helix.HelixAdmin;
+import org.apache.helix.manager.zk.ZKHelixAdmin;
+import org.apache.helix.model.InstanceConfig;
+import org.apache.s4.comm.tools.TaskSetup;
+import org.apache.s4.tools.S4ArgsBase;
+import org.apache.s4.tools.Tools;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.beust.jcommander.Parameter;
+import com.beust.jcommander.Parameters;
+
+public class AddNodes {
+
+    static Logger logger = LoggerFactory.getLogger(AddNodes.class);
+
+    public static void main(String[] args) {
+
+        ZKServerArgs clusterArgs = new ZKServerArgs();
+        Tools.parseArgs(clusterArgs, args);
+        try {
+
+            logger.info("Adding new nodes [{}] to cluster [{}] node(s)", clusterArgs.nbNodes, clusterArgs.clusterName);
+            HelixAdmin helixAdmin = new ZKHelixAdmin(clusterArgs.zkConnectionString);
+            int initialPort = clusterArgs.firstListeningPort;
+            if (clusterArgs.nbNodes > 0) {
+                String[] split = clusterArgs.nodes.split(",");
+                for (int i = 0; i < clusterArgs.nbNodes; i++) {
+                    InstanceConfig instanceConfig = new InstanceConfig("localhost_" + initialPort);
+                    String host = "localhost";
+                    if (split.length > 0 && split.length == clusterArgs.nbNodes) {
+                        host = split[i].trim();
+                    }
+                    instanceConfig.setHostName(host);
+                    instanceConfig.setPort("" + initialPort);
+                    instanceConfig.getRecord().setSimpleField("GROUP", clusterArgs.nodeGroup);
+                    helixAdmin.addInstance(clusterArgs.clusterName, instanceConfig);
+                    initialPort = initialPort + 1;
+                }
+            }
+            logger.info("New nodes configuration uploaded into zookeeper");
+        } catch (Exception e) {
+            logger.error("Cannot initialize zookeeper with specified configuration", e);
+        }
+
+    }
+
+    @Parameters(commandNames = "s4 addNodes", separators = "=", commandDescription = "Setup new S4 logical cluster")
+    static class ZKServerArgs extends S4ArgsBase {
+
+        @Parameter(names = { "-c", "-cluster" }, description = "S4 cluster name", required = true)
+        String clusterName = "s4-test-cluster";
+
+        @Parameter(names = "-nbNodes", description = "number of S4 nodes for the cluster", required = true)
+        int nbNodes = 1;
+
+        @Parameter(names = "-nodes", description = "Host names of the nodes", required = false)
+        String nodes = "";
+
+        @Parameter(names = "-zk", description = "Zookeeper connection string")
+        String zkConnectionString = "localhost:2181";
+
+        @Parameter(names = { "-flp", "-firstListeningPort" }, description = "Initial listening port for nodes in this cluster. First node listens on the specified port, other nodes listen on port initial + nodeIndex", required = true)
+        int firstListeningPort = -1;
+
+        @Parameter(names = { "-ng", "-nodeGroup" }, description = "Assign the nodes to one or more groups. This will be useful when you create task", required = false)
+        String nodeGroup = "default";
+    }
+
+}


Mime
View raw message