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
index c9cfcb7..cf2cd40 100644
--- 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
@@ -33,7 +33,7 @@
 import com.google.inject.assistedinject.FactoryModuleBuilder;
 import com.google.inject.name.Names;
 
-public class HelixBasedCommModule extends AbstractModule{
+public class HelixBasedCommModule extends AbstractModule {
 
     private static Logger logger = LoggerFactory.getLogger(DefaultCommModule.class);
     InputStream commConfigInputStream;
@@ -74,7 +74,8 @@
 
         // 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(StateModelFactory.class).annotatedWith(Names.named("s4.task.statemodelfactory")).to(
+                TaskStateModelFactory.class);
         bind(Assignment.class).to(AssignmentFromHelix.class);
         bind(Cluster.class).to(ClusterFromHelix.class);
 
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/RemoteEmitterFactory.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/RemoteEmitterFactory.java
index 5d79859..4c89aab 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/RemoteEmitterFactory.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/RemoteEmitterFactory.java
@@ -24,7 +24,7 @@
 /**
  * Used for creating RemoteEmitter instances depending on the cluster configuration. Follows the "assisted injection"
  * pattern from Guice 3.
- *
+ * 
  */
 public interface RemoteEmitterFactory {
 
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
index 9eebd06..ea35bd6 100644
--- 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
@@ -10,52 +10,45 @@
 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);
+public class S4StateModel extends StateModel {
+    private static Logger logger = LoggerFactory.getLogger(S4StateModel.class);
 
-  private final String streamName;
-  private final String partitionId;
+    private final String streamName;
+    private final String partitionId;
 
-  public S4StateModel(String partitionName)
-  {
-    String[] parts = partitionName.split("_");
-    this.streamName = parts[0];
-    this.partitionId = parts[1];
-  }
+    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 = "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 = "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 = "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());
-  }
+    @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());
+    }
 
 }
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
index eef64ea..3b885a8 100644
--- 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
@@ -2,8 +2,7 @@
 
 import org.apache.helix.participant.statemachine.StateModelFactory;
 
-
-public class TaskStateModelFactory extends StateModelFactory<S4StateModel>{
+public class TaskStateModelFactory extends StateModelFactory<S4StateModel> {
 
     @Override
     public S4StateModel createNewStateModel(String partitionName) {
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 a0e5002..120eee3 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
@@ -65,8 +65,7 @@
  */
 
 public class TCPEmitter implements Emitter, ClusterChangeListener {
-    private static final Logger logger = LoggerFactory
-            .getLogger(TCPEmitter.class);
+    private static final Logger logger = LoggerFactory.getLogger(TCPEmitter.class);
 
     private final int nettyTimeout;
 
@@ -87,24 +86,21 @@
     private final Lock lock;
 
     @Inject
-    SerializerDeserializer serDeser= new KryoSerDeser();
+    SerializerDeserializer serDeser = new KryoSerDeser();
 
     @Inject
-    public TCPEmitter(Cluster topology, @Named("s4.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();
 
         // Initialize data structures
-        //int clusterSize = this.topology.getPhysicalCluster().getNodes().size();
+        // int clusterSize = this.topology.getPhysicalCluster().getNodes().size();
         // TODO cluster can grow in size
-        nodeChannelMap = Maps.synchronizedBiMap(HashBiMap
-                .<InstanceConfig, Channel> create());
+        nodeChannelMap = Maps.synchronizedBiMap(HashBiMap.<InstanceConfig, Channel> create());
 
         // Initialize netty related structures
-        ChannelFactory factory = new NioClientSocketChannelFactory(
-                Executors.newCachedThreadPool(),
+        ChannelFactory factory = new NioClientSocketChannelFactory(Executors.newCachedThreadPool(),
                 Executors.newCachedThreadPool());
         bootstrap = new ClientBootstrap(factory);
         bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
@@ -137,19 +133,16 @@
         }
 
         try {
-            ChannelFuture connectFuture = this.bootstrap
-                    .connect(new InetSocketAddress(config.getHostName(),
-                            Integer.parseInt(config.getPort())));
+            ChannelFuture connectFuture = this.bootstrap.connect(new InetSocketAddress(config.getHostName(), Integer
+                    .parseInt(config.getPort())));
             connectFuture.await();
             if (connectFuture.isSuccess()) {
                 channels.add(connectFuture.getChannel());
-                nodeChannelMap
-                        .forcePut(config, connectFuture.getChannel());
+                nodeChannelMap.forcePut(config, connectFuture.getChannel());
                 return true;
             }
         } catch (InterruptedException ie) {
-            logger.error(String.format("Interrupted while connecting to %s:%d",
-                    config.getHostName(), config.getPort()));
+            logger.error(String.format("Interrupted while connecting to %s:%d", config.getHostName(), config.getPort()));
             Thread.currentThread().interrupt();
         }
         return false;
@@ -158,8 +151,7 @@
     private void sendMessage(String streamName, int partitionId, byte[] message) {
         ChannelBuffer buffer = ChannelBuffers.buffer(message.length);
         buffer.writeBytes(message);
-        InstanceConfig config = topology
-                .getDestination(streamName, partitionId);
+        InstanceConfig config = topology.getDestination(streamName, partitionId);
         if (!nodeChannelMap.containsKey(config)) {
             if (!connectTo(config)) {
                 // Couldn't connect, discard message
@@ -176,8 +168,7 @@
 
     @Override
     public boolean send(int partitionId, EventMessage message) {
-        sendMessage(message.getStreamName(), partitionId,
-                serDeser.serialize(message));
+        sendMessage(message.getStreamName(), partitionId, serDeser.serialize(message));
         return true;
     }
 
@@ -188,8 +179,7 @@
         }
         c.close().addListener(new ChannelFutureListener() {
             @Override
-            public void operationComplete(ChannelFuture future)
-                    throws Exception {
+            public void operationComplete(ChannelFuture future) throws Exception {
                 if (future.isSuccess())
                     channels.remove(future.getChannel());
                 else
@@ -208,18 +198,18 @@
         }
     }
 
-    //@Override
+    // @Override
     public int getPartitionCount() {
         return topology.getPhysicalCluster().getPartitionCount();
     }
+
     public int getPartitionCount(String streamName) {
-    return topology.getPhysicalCluster().getPartitionCount();
-  }
+        return topology.getPhysicalCluster().getPartitionCount();
+    }
 
     class ExceptionHandler extends SimpleChannelUpstreamHandler {
         @Override
-        public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e)
-                throws Exception {
+        public void exceptionCaught(ChannelHandlerContext ctx, ExceptionEvent e) throws Exception {
             Throwable t = e.getCause();
             if (t instanceof ClosedChannelException) {
                 nodeChannelMap.inverse().remove(e.getChannel());
@@ -248,10 +238,8 @@
                 try {
                     // TODO handle possible cluster reconfiguration between send
                     // and failure callback
-                    logger.warn(
-                            "Failed to send message to node {} (according to current cluster information)",
-                            topology.getPhysicalCluster().getNodes()
-                                    .get(partitionId));
+                    logger.warn("Failed to send message to node {} (according to current cluster information)",
+                            topology.getPhysicalCluster().getNodes().get(partitionId));
                 } catch (IndexOutOfBoundsException ignored) {
                     // cluster was changed
                 }
@@ -262,6 +250,6 @@
 
     @Override
     public void onChange() {
-        
+
     }
 }
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tools/TaskSetup.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tools/TaskSetup.java
index 952306a..dfe98bf 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tools/TaskSetup.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/tools/TaskSetup.java
@@ -34,100 +34,80 @@
  * Used for defining and dimensioning logical clusters in Zookeeper.
  * 
  */
-public class TaskSetup
-{
+public class TaskSetup {
 
-  private ZkClient zkclient;
-  private boolean isHelixEnabled = true;
-  private HelixAdmin helixAdmin;
-  org.apache.helix.manager.zk.ZkClient helixZkClient;
+    private ZkClient zkclient;
+    private boolean isHelixEnabled = true;
+    private HelixAdmin helixAdmin;
+    org.apache.helix.manager.zk.ZkClient helixZkClient;
 
-  public TaskSetup(String zookeeperAddress)
-  {
-    if (isHelixEnabled)
-    {
-      helixAdmin = new ZKHelixAdmin(zookeeperAddress);
-    } else
-    {
-      zkclient = new ZkClient(zookeeperAddress);
-      zkclient.setZkSerializer(new ZNRecordSerializer());
-      if (!zkclient.waitUntilConnected(10, TimeUnit.SECONDS))
-      {
-        throw new RuntimeException(
-            "Could not connect to ZooKeeper after 10 seconds.");
-      }
-    }
-  }
-
-  public void clean(String clusterName)
-  {
-    if (isHelixEnabled)
-    {
-      helixAdmin.dropCluster(clusterName);
-    } else
-    {
-      zkclient.deleteRecursive("/s4/clusters/" + clusterName);
-    }
-  }
-
-  public void setup(String cluster, int tasks, int initialPort)
-  {
-    if (isHelixEnabled)
-    {
-      helixAdmin.addCluster(cluster, false);
-      StateModelDefinition onlineofflinemodel = new StateModelDefinition(
-          new StateModelConfigGenerator().generateConfigForOnlineOffline());
-      StateModelDefinition leaderstandbymodel = new StateModelDefinition(
-          new StateModelConfigGenerator().generateConfigForLeaderStandby());
-
-      helixAdmin.addStateModelDef(cluster, "OnlineOffline", onlineofflinemodel);
-      helixAdmin.addStateModelDef(cluster, "LeaderStandby", leaderstandbymodel);
-      
-      for (int i = 0; i < tasks; i++)
-      {
-        InstanceConfig instanceConfig = new InstanceConfig("localhost_"
-            + initialPort);
-        instanceConfig.setHostName("localhost");
-        instanceConfig.setPort("" + initialPort);
-        helixAdmin.addInstance(cluster, instanceConfig);
-        initialPort = initialPort + 1;
-      }
-
-      return;
-    }
-    try
-    {
-      zkclient.createPersistent("/s4/streams", true);
-    } catch (ZkException ignored)
-    {
-      // ignore if exists
+    public TaskSetup(String zookeeperAddress) {
+        if (isHelixEnabled) {
+            helixAdmin = new ZKHelixAdmin(zookeeperAddress);
+        } else {
+            zkclient = new ZkClient(zookeeperAddress);
+            zkclient.setZkSerializer(new ZNRecordSerializer());
+            if (!zkclient.waitUntilConnected(10, TimeUnit.SECONDS)) {
+                throw new RuntimeException("Could not connect to ZooKeeper after 10 seconds.");
+            }
+        }
     }
 
-    zkclient.createPersistent("/s4/clusters/" + cluster + "/tasks", true);
-    zkclient.createPersistent("/s4/clusters/" + cluster + "/process", true);
-    zkclient.createPersistent("/s4/clusters/" + cluster + "/app", true);
-    for (int i = 0; i < tasks; i++)
-    {
-      String taskId = "Task-" + i;
-      ZNRecord record = new ZNRecord(taskId);
-      record.putSimpleField("taskId", taskId);
-      record.putSimpleField("port", String.valueOf(initialPort + i));
-      record.putSimpleField("partition", String.valueOf(i));
-      record.putSimpleField("cluster", cluster);
-      zkclient.createPersistent("/s4/clusters/" + cluster + "/tasks/" + taskId,
-          record);
+    public void clean(String clusterName) {
+        if (isHelixEnabled) {
+            helixAdmin.dropCluster(clusterName);
+        } else {
+            zkclient.deleteRecursive("/s4/clusters/" + clusterName);
+        }
     }
-  }
 
-  public void disconnect()
-  {
-    if (isHelixEnabled)
-    {
-      helixZkClient.close();
-    } else
-    {
-      zkclient.close();
+    public void setup(String cluster, int tasks, int initialPort) {
+        if (isHelixEnabled) {
+            helixAdmin.addCluster(cluster, false);
+            StateModelDefinition onlineofflinemodel = new StateModelDefinition(
+                    new StateModelConfigGenerator().generateConfigForOnlineOffline());
+            StateModelDefinition leaderstandbymodel = new StateModelDefinition(
+                    new StateModelConfigGenerator().generateConfigForLeaderStandby());
+
+            helixAdmin.addStateModelDef(cluster, "OnlineOffline", onlineofflinemodel);
+            helixAdmin.addStateModelDef(cluster, "LeaderStandby", leaderstandbymodel);
+
+            for (int i = 0; i < tasks; i++) {
+                InstanceConfig instanceConfig = new InstanceConfig("localhost_" + initialPort);
+                instanceConfig.setHostName("localhost");
+                instanceConfig.setPort("" + initialPort);
+                helixAdmin.addInstance(cluster, instanceConfig);
+                initialPort = initialPort + 1;
+            }
+
+            return;
+        }
+        try {
+            zkclient.createPersistent("/s4/streams", true);
+        } catch (ZkException ignored) {
+            // ignore if exists
+        }
+
+        zkclient.createPersistent("/s4/clusters/" + cluster + "/tasks", true);
+        zkclient.createPersistent("/s4/clusters/" + cluster + "/process", true);
+        zkclient.createPersistent("/s4/clusters/" + cluster + "/app", true);
+        for (int i = 0; i < tasks; i++) {
+            String taskId = "Task-" + i;
+            ZNRecord record = new ZNRecord(taskId);
+            record.putSimpleField("taskId", taskId);
+            record.putSimpleField("port", String.valueOf(initialPort + i));
+            record.putSimpleField("partition", String.valueOf(i));
+            record.putSimpleField("cluster", cluster);
+            zkclient.createPersistent("/s4/clusters/" + cluster + "/tasks/" + taskId, record);
+        }
     }
-  }
+
+    public void disconnect() {
+        if (isHelixEnabled) {
+            helixZkClient.close();
+        } else {
+            zkclient.close();
+        }
+    }
 
 }
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
index 7e0b4ac..243faac 100644
--- 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
@@ -35,135 +35,109 @@
 import com.google.inject.name.Named;
 
 @Singleton
-public class AssignmentFromHelix implements Assignment
-{
-  private static final Logger logger = LoggerFactory
-      .getLogger(AssignmentFromHelix.class);
+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 String clusterName;
+    private final String zookeeperAddress;
+    private String machineId;
+    private HelixManager zkHelixManager;
 
-  private final StateModelFactory<? extends StateModel> taskStateModelFactory;
-  //private final StateModelFactory<? extends StateModel> appStateModelFactory;
+    private HelixDataAccessor helixDataAccessor;
+    AtomicReference<ClusterNode> clusterNodeRef;
+    private final Lock lock;
+    private final AtomicBoolean currentlyOwningTask;
+    private final Condition taskAcquired;
 
-  @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);
-  }
+    private final StateModelFactory<? extends StateModel> taskStateModelFactory;
 
-  @Inject
-  public void init()
-  {
-    //joinCluster();
-  }
+    // private final StateModelFactory<? extends StateModel> appStateModelFactory;
 
-  @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;
-          }
+    @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";
         }
-        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();
+        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();
+        }
+    }
 
 }
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 0b81854..c482624 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
@@ -44,7 +44,7 @@
 
 /**
  * Handles partition assignment through Zookeeper.
- *
+ * 
  */
 @Singleton
 public class AssignmentFromZK implements Assignment, IZkChildListener, IZkStateListener, IZkDataListener {
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
index 3226fab..19e0628 100644
--- 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
@@ -46,172 +46,145 @@
 import org.apache.helix.spectator.RoutingTableProvider;
 
 /**
- * Represents a logical cluster definition fetched from Zookeeper. Notifies
- * listeners of runtime changes in the configuration.
+ * Represents a logical cluster definition fetched from Zookeeper. Notifies listeners of runtime changes in the
+ * configuration.
  * 
  */
-public class ClusterFromHelix extends RoutingTableProvider implements Cluster
-{
+public class ClusterFromHelix extends RoutingTableProvider implements Cluster {
 
-  private static Logger logger = LoggerFactory
-      .getLogger(ClusterFromHelix.class);
+    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;
+    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();
+    /**
+     * 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();
+    /**
+     * 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());
+    @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();
         }
-      }
-      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 PhysicalCluster getPhysicalCluster() {
+        return clusterRef.get();
     }
-  }
-  
-  @Override
-  public Integer getPartitionCount(String streamName){
-    Integer numPartitions = partitionCountMapRef.get().get(streamName);
-    if(numPartitions==null){
-      return -1;
+
+    @Override
+    public void addListener(ClusterChangeListener listener) {
+        logger.info("Adding topology change listener:" + listener);
+        listeners.add(listener);
     }
-    return numPartitions;
-  }
+
+    @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;
+    }
 
 }
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 f6d4df9..3e4567a 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
@@ -42,7 +42,7 @@
 /**
  * Represents a logical cluster definition fetched from Zookeeper. Notifies listeners of runtime changes in the
  * configuration.
- *
+ * 
  */
 public class ClusterFromZK implements Cluster, IZkChildListener, IZkDataListener, IZkStateListener {
 
@@ -216,15 +216,13 @@
     }
 
     @Override
-    public InstanceConfig getDestination(String streamName, int partitionId)
-    {
-      return null;
+    public InstanceConfig getDestination(String streamName, int partitionId) {
+        return null;
     }
 
     @Override
-    public Integer getPartitionCount(String streamName)
-    {
-      return null;
+    public Integer getPartitionCount(String streamName) {
+        return null;
     }
 
 }
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ClusterNode.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ClusterNode.java
index e622107..1e117c8 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ClusterNode.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ClusterNode.java
@@ -20,7 +20,7 @@
 
 /**
  * Represents an node.
- *
+ * 
  */
 public class ClusterNode {
     private int partition;
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
index 217d4e7..4e5351b 100644
--- 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
@@ -55,11 +55,8 @@
         this.clusterName = clusterName;
         this.connectionTimeout = connectionTimeout;
 
-
     }
 
-  
-
     public Cluster getCluster(String clusterName) {
         return clusters.get(clusterName);
     }
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 ce53c1a..cbf209e 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
@@ -99,8 +99,9 @@
     }
 
     public int getPartitionCount(String stream) {
-      return numPartitions;
-  }
+        return numPartitions;
+    }
+
     /**
      * @param node
      */
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 7720bbf..7b75ff4 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
@@ -107,7 +107,9 @@
 
     }
 
-    /* (non-Javadoc)
+    /*
+     * (non-Javadoc)
+     * 
      * @see org.apache.s4.comm.topology.RemoteStreamsManager#getConsumers(java.lang.String)
      */
     @Override
@@ -186,8 +188,11 @@
         streams.get(streamName).put(type.getCollectionName(), Collections.unmodifiableSet(consumers));
     }
 
-    /* (non-Javadoc)
-     * @see org.apache.s4.comm.topology.RemoteStreamsManager#addOutputStream(java.lang.String, java.lang.String, java.lang.String)
+    /*
+     * (non-Javadoc)
+     * 
+     * @see org.apache.s4.comm.topology.RemoteStreamsManager#addOutputStream(java.lang.String, java.lang.String,
+     * java.lang.String)
      */
     @Override
     public void addOutputStream(String appId, String clusterName, String streamName) {
@@ -219,7 +224,9 @@
         zkClient.createPersistent(StreamType.CONSUMER.getPath(streamName), true);
     }
 
-    /* (non-Javadoc)
+    /*
+     * (non-Javadoc)
+     * 
      * @see org.apache.s4.comm.topology.RemoteStreamsManager#addInputStream(int, java.lang.String, java.lang.String)
      */
     @Override
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/RemoteStreamsManager.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/RemoteStreamsManager.java
index 94c95b6..3eb6d6a 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/RemoteStreamsManager.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/RemoteStreamsManager.java
@@ -2,22 +2,19 @@
 
 import java.util.Set;
 
-public interface RemoteStreamsManager
-{
+public interface RemoteStreamsManager {
 
-  public abstract Set<StreamConsumer> getConsumers(String streamName);
+    public abstract Set<StreamConsumer> getConsumers(String streamName);
 
-  public abstract void addOutputStream(String appId, String clusterName,
-      String streamName);
+    public abstract void addOutputStream(String appId, String clusterName, String streamName);
 
-  /**
-   * Publishes interest in a stream from an application.
-   * 
-   * @param appId
-   * @param clusterName
-   * @param streamName
-   */
-  public abstract void addInputStream(int appId, String clusterName,
-      String streamName);
+    /**
+     * Publishes interest in a stream from an application.
+     * 
+     * @param appId
+     * @param clusterName
+     * @param streamName
+     */
+    public abstract void addInputStream(int appId, String clusterName, String streamName);
 
-}
\ No newline at end of file
+}
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/RemoteStreamsManagerImpl.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/RemoteStreamsManagerImpl.java
index 957318d..845e649 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/RemoteStreamsManagerImpl.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/RemoteStreamsManagerImpl.java
@@ -2,26 +2,21 @@
 
 import java.util.Set;
 
-public class RemoteStreamsManagerImpl implements RemoteStreamsManager
-{
+public class RemoteStreamsManagerImpl implements RemoteStreamsManager {
 
-  @Override
-  public Set<StreamConsumer> getConsumers(String streamName)
-  {
-    return null;
-  }
+    @Override
+    public Set<StreamConsumer> getConsumers(String streamName) {
+        return null;
+    }
 
-  @Override
-  public void addOutputStream(String appId, String clusterName,
-      String streamName)
-  {
-    
-  }
+    @Override
+    public void addOutputStream(String appId, String clusterName, String streamName) {
 
-  @Override
-  public void addInputStream(int appId, String clusterName, String streamName)
-  {
-    
-  }
+    }
+
+    @Override
+    public void addInputStream(int appId, String clusterName, String streamName) {
+
+    }
 
 }
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ZNRecordSerializer.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ZNRecordSerializer.java
index b55f469..9858aef 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ZNRecordSerializer.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ZNRecordSerializer.java
@@ -24,12 +24,12 @@
 import com.google.common.base.Preconditions;
 import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
+
 /**
  * 
  * Utility to serialize/deserialize data in ZK. <br/>
- * Using Json format and Gson library. 
- * TODO: Explore other libraries like jackson much richer features.
- * Gson needs no-arg constructor to work with without additional work
+ * Using Json format and Gson library. TODO: Explore other libraries like jackson much richer features. Gson needs
+ * no-arg constructor to work with without additional work
  */
 public class ZNRecordSerializer implements ZkSerializer {
 
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ZkClient.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ZkClient.java
index e5a8689..cab6c0f 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ZkClient.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/topology/ZkClient.java
@@ -25,72 +25,70 @@
 import org.I0Itec.zkclient.exception.ZkNoNodeException;
 import org.I0Itec.zkclient.serialize.ZkSerializer;
 import org.apache.zookeeper.data.Stat;
+
 /**
  * 
  * Overwriting the ZKclient since the org.I0Itec.zkclient.ZkClient does not expose some important methods
  */
 public class ZkClient extends org.I0Itec.zkclient.ZkClient {
 
-	public ZkClient(IZkConnection connection, int connectionTimeout,
-			ZkSerializer zkSerializer) {
-		super(connection, connectionTimeout, zkSerializer);
-	}
+    public ZkClient(IZkConnection connection, int connectionTimeout, ZkSerializer zkSerializer) {
+        super(connection, connectionTimeout, zkSerializer);
+    }
 
-	public ZkClient(IZkConnection connection, int connectionTimeout) {
-		super(connection, connectionTimeout);
-	}
+    public ZkClient(IZkConnection connection, int connectionTimeout) {
+        super(connection, connectionTimeout);
+    }
 
-	public ZkClient(IZkConnection connection) {
-		super(connection);
-	}
+    public ZkClient(IZkConnection connection) {
+        super(connection);
+    }
 
-	public ZkClient(String zkServers, int sessionTimeout, int connectionTimeout) {
-		super(zkServers, sessionTimeout, connectionTimeout);
-	}
+    public ZkClient(String zkServers, int sessionTimeout, int connectionTimeout) {
+        super(zkServers, sessionTimeout, connectionTimeout);
+    }
 
-	public ZkClient(String zkServers, int connectionTimeout) {
-		super(zkServers, connectionTimeout);
-	}
+    public ZkClient(String zkServers, int connectionTimeout) {
+        super(zkServers, connectionTimeout);
+    }
 
-	public ZkClient(String serverstring) {
-		super(serverstring);
-	}
+    public ZkClient(String serverstring) {
+        super(serverstring);
+    }
 
-	public IZkConnection getConnection() {
-		return _connection;
-	}
-	
-	public long getSessionId(){
-		return ((ZkConnection)_connection).getZookeeper().getSessionId();
-	}
+    public IZkConnection getConnection() {
+        return _connection;
+    }
 
-	public Stat getStat(final String path) {
-		Stat stat = retryUntilConnected(new Callable<Stat>() {
+    public long getSessionId() {
+        return ((ZkConnection) _connection).getZookeeper().getSessionId();
+    }
 
-			@Override
-			public Stat call() throws Exception {
-				Stat stat = ((ZkConnection) _connection).getZookeeper().exists(
-						path, false);
-				return stat;
-			}
-		});
+    public Stat getStat(final String path) {
+        Stat stat = retryUntilConnected(new Callable<Stat>() {
 
-		return stat;
-	}
+            @Override
+            public Stat call() throws Exception {
+                Stat stat = ((ZkConnection) _connection).getZookeeper().exists(path, false);
+                return stat;
+            }
+        });
 
-	@SuppressWarnings("unchecked")
-	@Override
-	public <T extends Object> T readData(String path,
-			boolean returnNullIfPathNotExists) {
-		T data = null;
-		try {
-			data = (T) readData(path, null);
-		} catch (ZkNoNodeException e) {
-			if (!returnNullIfPathNotExists) {
-				throw e;
-			}
-		}
-		return data;
-	}
+        return stat;
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override
+    public <T extends Object> T readData(String path, boolean returnNullIfPathNotExists) {
+        T data = null;
+        try {
+            data = (T) readData(path, null);
+        } catch (ZkNoNodeException e) {
+            if (!returnNullIfPathNotExists) {
+                throw e;
+            }
+        }
+        return data;
+    }
 
 }
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 fb18dd6..a326535 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
@@ -39,7 +39,7 @@
 
 /**
  * UDP based emitter.
- *
+ * 
  */
 public class UDPEmitter implements Emitter, ClusterChangeListener {
     private DatagramSocket socket;
@@ -103,11 +103,11 @@
         return true;
     }
 
-   // @Override
+    // @Override
     public int getPartitionCount() {
         return topology.getPhysicalCluster().getPartitionCount();
     }
-    
+
     @Override
     public void onChange() {
         refreshCluster();
@@ -131,8 +131,7 @@
     }
 
     @Override
-    public int getPartitionCount(String stream)
-    {
-      return topology.getPhysicalCluster().getPartitionCount(stream);      
+    public int getPartitionCount(String stream) {
+        return topology.getPhysicalCluster().getPartitionCount(stream);
     }
 }
diff --git a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/udp/UDPRemoteEmitter.java b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/udp/UDPRemoteEmitter.java
index 050a6ec..da015d7 100644
--- a/subprojects/s4-comm/src/main/java/org/apache/s4/comm/udp/UDPRemoteEmitter.java
+++ b/subprojects/s4-comm/src/main/java/org/apache/s4/comm/udp/UDPRemoteEmitter.java
@@ -25,7 +25,7 @@
 
 /**
  * UDP-based emitter for sending events to remote clusters.
- *
+ * 
  */
 public class UDPRemoteEmitter extends UDPEmitter {
 
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 97ec9af..1c07870 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
@@ -83,14 +83,14 @@
         /* Use Kryo to serialize events. */
         bind(SerializerDeserializer.class).to(KryoSerDeser.class);
 
-        bind(StateModelFactory.class).annotatedWith(Names.named("s4.app.statemodelfactory")).to(AppStateModelFactory.class);
+        bind(StateModelFactory.class).annotatedWith(Names.named("s4.app.statemodelfactory")).to(
+                AppStateModelFactory.class);
 
         bind(DeploymentManager.class).to(HelixBasedDeploymentManager.class);
-        
-        bind(RemoteSendersManager.class).to(RemoteSendersManagerImpl.class);
-        
-        bind(RemoteStreamsManager.class).to(RemoteStreamsManagerImpl.class);
 
+        bind(RemoteSendersManager.class).to(RemoteSendersManagerImpl.class);
+
+        bind(RemoteStreamsManager.class).to(RemoteStreamsManagerImpl.class);
 
         bind(S4RLoaderFactory.class);
 
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/Main.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/Main.java
index 810b8f4..9ae356b 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/core/Main.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/Main.java
@@ -155,11 +155,12 @@
                 }
                 combinedModule = Modules.override(combinedModule).with(new ParametersInjectionModule(namedParameters));
             }
-            
+
             injector = Guice.createInjector(combinedModule);
-            //start a HelixController to manage the cluster
+            // start a HelixController to manage the cluster
             String controllerName = Inet4Address.getLocalHost().getCanonicalHostName() + UUID.randomUUID().toString();
-            HelixControllerMain.startHelixController(mainArgs.zkConnectionString, mainArgs.clusterName, controllerName, HelixControllerMain.STANDALONE);
+            HelixControllerMain.startHelixController(mainArgs.zkConnectionString, mainArgs.clusterName, controllerName,
+                    HelixControllerMain.STANDALONE);
 
             if (mainArgs.appClass != null) {
                 logger.info("Starting S4 node with single application from class [{}]", mainArgs.appClass);
@@ -190,7 +191,7 @@
 
         @Parameter(names = { "-c", "-cluster" }, description = "cluster name", required = true)
         String clusterName = null;
-        
+
         @Parameter(names = { "-id", "-nodeId" }, description = "Node/Instance id that uniquely identifies a node", required = false)
         String instanceName = null;
 
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/RemoteSender.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/RemoteSender.java
index 6bc52e3..c8da41a 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/core/RemoteSender.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/RemoteSender.java
@@ -24,7 +24,7 @@
 
 /**
  * Sends events to a remote cluster.
- *
+ * 
  */
 public class RemoteSender {
 
@@ -41,7 +41,8 @@
     public void send(String hashKey, EventMessage eventMessage) {
         if (hashKey == null) {
             // round robin by default
-            emitter.send(Math.abs(targetPartition++ % emitter.getPartitionCount(eventMessage.getStreamName())), eventMessage);
+            emitter.send(Math.abs(targetPartition++ % emitter.getPartitionCount(eventMessage.getStreamName())),
+                    eventMessage);
         } else {
             int partition = (int) (hasher.hash(hashKey) % emitter.getPartitionCount(eventMessage.getStreamName()));
             emitter.send(partition, eventMessage);
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/RemoteSenders.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/RemoteSenders.java
index a43aa7d..e8ed8e0 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/core/RemoteSenders.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/RemoteSenders.java
@@ -38,7 +38,7 @@
 /**
  * Sends events to remote clusters. Target clusters are selected dynamically based on the stream name information from
  * the event.
- *
+ * 
  */
 public class RemoteSenders implements RemoteSendersManager {
 
@@ -61,7 +61,9 @@
 
     Map<String, RemoteSender> sendersByTopology = new HashMap<String, RemoteSender>();
 
-    /* (non-Javadoc)
+    /*
+     * (non-Javadoc)
+     * 
      * @see org.apache.s4.core.RemoteSendersManager#send(java.lang.String, org.apache.s4.base.Event)
      */
     @Override
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/RemoteSendersManager.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/RemoteSendersManager.java
index 3e0e462..0f1f9fa 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/core/RemoteSendersManager.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/RemoteSendersManager.java
@@ -2,9 +2,8 @@
 
 import org.apache.s4.base.Event;
 
-public interface RemoteSendersManager
-{
+public interface RemoteSendersManager {
 
-  public abstract void send(String hashKey, Event event);
+    public abstract void send(String hashKey, Event event);
 
-}
\ No newline at end of file
+}
diff --git a/subprojects/s4-core/src/main/java/org/apache/s4/core/RemoteSendersManagerImpl.java b/subprojects/s4-core/src/main/java/org/apache/s4/core/RemoteSendersManagerImpl.java
index 13f8ac8..ed2ce4a 100644
--- a/subprojects/s4-core/src/main/java/org/apache/s4/core/RemoteSendersManagerImpl.java
+++ b/subprojects/s4-core/src/main/java/org/apache/s4/core/RemoteSendersManagerImpl.java
@@ -2,13 +2,11 @@
 
 import org.apache.s4.base.Event;
 
-public class RemoteSendersManagerImpl implements RemoteSendersManager
-{
+public class RemoteSendersManagerImpl implements RemoteSendersManager {
 
-  @Override
-  public void send(String hashKey, Event event)
-  {
-    
-  }
+    @Override
+    public void send(String hashKey, Event event) {
+
+    }
 
 }
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
index 38f71a8..d47ee0b 100644
--- 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
@@ -23,90 +23,69 @@
 import com.google.common.io.Files;
 
 @StateModelInfo(states = { "ONLINE,OFFLINE" }, initialState = "OFFLINE")
-public class AppStateModel extends StateModel
-{
-  private static Logger logger = LoggerFactory.getLogger(AppStateModel.class);
-  private final String appName;
-  private final Server server;
+public class AppStateModel extends StateModel {
+    private static Logger logger = LoggerFactory.getLogger(AppStateModel.class);
+    private final String appName;
+    private final Server server;
 
-  public AppStateModel(Server server, String appName)
-  {
-    this.server = server;
-    this.appName = appName;
-  }
-
-  @Transition(from = "OFFLINE", to = "ONLINE")
-  public void deploy(Message message, NotificationContext context) throws Exception
-  {
-    HelixManager manager = context.getManager();
-    ConfigAccessor configAccessor = manager.getConfigAccessor();
-    ConfigScopeBuilder builder = new ConfigScopeBuilder();
-    ConfigScope scope = builder.forCluster(manager.getClusterName()).forResource(appName).build();
-    String uriString = configAccessor.get(scope,
-        DistributedDeploymentManager.S4R_URI);
-    String clusterName = manager.getClusterName();
-    try
-    {
-      URI uri = new URI(uriString);
-      // 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 [{}]",
-                appName, e1.getClass().getName() + "->" + e1.getMessage());
-        throw new DeploymentFailedException("Cannot deploy application ["
-            + appName + "]", e1);
-      }
-      localS4RFileCopy.deleteOnExit();
-      try
-      {
-        if (ByteStreams.copy(DistributedDeploymentManager.fetchS4App(uri),
-            Files.newOutputStreamSupplier(localS4RFileCopy)) == 0)
-        {
-          throw new DeploymentFailedException("Cannot copy archive from ["
-              + uri.toString() + "] to [" + localS4RFileCopy.getAbsolutePath()
-              + "] (nothing was copied)");
-        }
-      } catch (IOException e)
-      {
-        throw new DeploymentFailedException("Cannot deploy application ["
-            + appName + "] from URI [" + uri.toString() + "] ", e);
-      }
-      // install locally
-      App loaded = server.loadApp(localS4RFileCopy, appName);
-      if (loaded != null)
-      {
-        logger.info("Successfully installed application {}", appName);
-        // TODO sync with other nodes? (e.g. wait for other apps deployed before
-        // starting?
-        server.startApp(loaded, appName, clusterName);
-      } else
-      {
-        throw new DeploymentFailedException("Cannot deploy application ["
-            + appName + "] from URI [" + uri.toString()
-            + "] : cannot start application");
-      }
-
-    } catch (URISyntaxException e)
-    {
-      logger
-          .error(
-              "Cannot deploy app {} : invalid uri for fetching s4r archive {} : {} ",
-              new String[] { appName, uriString, e.getMessage() });
-      throw new DeploymentFailedException("Cannot deploy application ["
-          + appName + "]", e);
+    public AppStateModel(Server server, String appName) {
+        this.server = server;
+        this.appName = appName;
     }
-  }
 
-  @Transition(from = "OFFLINE", to = "ONLINE")
-  public void undeploy(Message message, NotificationContext context) throws Exception
-  {
-    //todo
-  }
+    @Transition(from = "OFFLINE", to = "ONLINE")
+    public void deploy(Message message, NotificationContext context) throws Exception {
+        HelixManager manager = context.getManager();
+        ConfigAccessor configAccessor = manager.getConfigAccessor();
+        ConfigScopeBuilder builder = new ConfigScopeBuilder();
+        ConfigScope scope = builder.forCluster(manager.getClusterName()).forResource(appName).build();
+        String uriString = configAccessor.get(scope, DistributedDeploymentManager.S4R_URI);
+        String clusterName = manager.getClusterName();
+        try {
+            URI uri = new URI(uriString);
+            // 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 [{}]",
+                        appName, e1.getClass().getName() + "->" + e1.getMessage());
+                throw new DeploymentFailedException("Cannot deploy application [" + appName + "]", e1);
+            }
+            localS4RFileCopy.deleteOnExit();
+            try {
+                if (ByteStreams.copy(DistributedDeploymentManager.fetchS4App(uri),
+                        Files.newOutputStreamSupplier(localS4RFileCopy)) == 0) {
+                    throw new DeploymentFailedException("Cannot copy archive from [" + uri.toString() + "] to ["
+                            + localS4RFileCopy.getAbsolutePath() + "] (nothing was copied)");
+                }
+            } catch (IOException e) {
+                throw new DeploymentFailedException("Cannot deploy application [" + appName + "] from URI ["
+                        + uri.toString() + "] ", e);
+            }
+            // install locally
+            App loaded = server.loadApp(localS4RFileCopy, appName);
+            if (loaded != null) {
+                logger.info("Successfully installed application {}", appName);
+                // TODO sync with other nodes? (e.g. wait for other apps deployed before
+                // starting?
+                server.startApp(loaded, appName, clusterName);
+            } else {
+                throw new DeploymentFailedException("Cannot deploy application [" + appName + "] from URI ["
+                        + uri.toString() + "] : cannot start application");
+            }
+
+        } catch (URISyntaxException e) {
+            logger.error("Cannot deploy app {} : invalid uri for fetching s4r archive {} : {} ", new String[] {
+                    appName, uriString, e.getMessage() });
+            throw new DeploymentFailedException("Cannot deploy application [" + appName + "]", e);
+        }
+    }
+
+    @Transition(from = "OFFLINE", to = "ONLINE")
+    public void undeploy(Message message, NotificationContext context) throws Exception {
+        // todo
+    }
 
 }
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
index 3af4531..ef6ae58 100644
--- 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
@@ -5,20 +5,20 @@
 
 import com.google.inject.Inject;
 import com.google.inject.Singleton;
+
 @Singleton
-public class AppStateModelFactory extends StateModelFactory<AppStateModel>
-{
-  private final Server server;
-  
-  @Inject
-  public AppStateModelFactory(Server server){
-    this.server = server;
-    
-  }
-  @Override
-  public AppStateModel createNewStateModel(String partitionName)
-  {
-    return new AppStateModel(server,partitionName);
-  }
+public class AppStateModelFactory extends StateModelFactory<AppStateModel> {
+    private final Server server;
+
+    @Inject
+    public AppStateModelFactory(Server server) {
+        this.server = server;
+
+    }
+
+    @Override
+    public AppStateModel createNewStateModel(String partitionName) {
+        return new AppStateModel(server, partitionName);
+    }
 
 }
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
index ed32085..5dcfa76 100644
--- 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
@@ -5,39 +5,24 @@
 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;
+public class HelixBasedDeploymentManager implements DeploymentManager {
+    private final Server server;
+    boolean deployed = false;
+    private final String clusterName;
 
-  @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)
-  {
-    this.clusterName = clusterName;
-    this.server = server;
+    @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) {
+        this.clusterName = clusterName;
+        this.server = server;
 
-  }
+    }
 
-  @Override
-  public void start()
-  {
-   /* File s4r = new File(
-        "/Users/kgopalak/Documents/projects/s4/incubator-s4-helix/myApp/build/libs/myApp.s4r");
-    String appName = "myApp";
-    try
-    {
-      App loaded = server.loadApp(s4r, "myApp");
-      server.startApp(loaded, appName, clusterName);
-    } catch (Exception e)
-    {
-      e.printStackTrace();
-    }*/
-  }
+    @Override
+    public void start() {
+        
+    }
 
 }
diff --git a/subprojects/s4-tools/src/main/java/org/apache/s4/tools/AddNodes.java b/subprojects/s4-tools/src/main/java/org/apache/s4/tools/AddNodes.java
index 5bab316..6dfc82a 100644
--- a/subprojects/s4-tools/src/main/java/org/apache/s4/tools/AddNodes.java
+++ b/subprojects/s4-tools/src/main/java/org/apache/s4/tools/AddNodes.java
@@ -38,7 +38,7 @@
         Tools.parseArgs(clusterArgs, args);
         try {
 
-            logger.info("Adding new nodes [{}] to cluster [{}] node(s)",  clusterArgs.nbNodes, clusterArgs.clusterName);
+            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) {
@@ -74,14 +74,14 @@
 
         @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)
+
+        @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";
     }
 
diff --git a/subprojects/s4-tools/src/main/java/org/apache/s4/tools/CreateCluster.java b/subprojects/s4-tools/src/main/java/org/apache/s4/tools/CreateCluster.java
index 38fbbe8..190c8c0 100644
--- a/subprojects/s4-tools/src/main/java/org/apache/s4/tools/CreateCluster.java
+++ b/subprojects/s4-tools/src/main/java/org/apache/s4/tools/CreateCluster.java
@@ -88,10 +88,10 @@
 
         @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 = "Name of the App", required = false, arity = 1)
         String nodeGroup = "default";
-        
+
     }
 
 }
diff --git a/subprojects/s4-tools/src/main/java/org/apache/s4/tools/DeployApp.java b/subprojects/s4-tools/src/main/java/org/apache/s4/tools/DeployApp.java
index ff1cb88..83a7088 100644
--- a/subprojects/s4-tools/src/main/java/org/apache/s4/tools/DeployApp.java
+++ b/subprojects/s4-tools/src/main/java/org/apache/s4/tools/DeployApp.java
@@ -47,14 +47,14 @@
         for (String instanceName : instancesInCluster) {
             InstanceConfig instanceConfig = admin.getInstanceConfig(deployArgs.clusterName, instanceName);
             String nodeGroup = instanceConfig.getRecord().getSimpleField("GROUP");
-            if(nodeGroup.equals(deployArgs.nodeGroup)){
+            if (nodeGroup.equals(deployArgs.nodeGroup)) {
                 instancesInGroup.add(instanceName);
             }
         }
-        for(String instanceName:instancesInGroup){
+        for (String instanceName : instancesInGroup) {
             is.setPartitionState(deployArgs.appName, instanceName, "ONLINE");
         }
-        
+
         admin.setResourceIdealState(deployArgs.clusterName, deployArgs.appName, is);
     }
 
diff --git a/subprojects/s4-tools/src/main/java/org/apache/s4/tools/GenericEventAdapter.java b/subprojects/s4-tools/src/main/java/org/apache/s4/tools/GenericEventAdapter.java
index b54187f..ce41ecd 100644
--- a/subprojects/s4-tools/src/main/java/org/apache/s4/tools/GenericEventAdapter.java
+++ b/subprojects/s4-tools/src/main/java/org/apache/s4/tools/GenericEventAdapter.java
@@ -19,64 +19,53 @@
 import com.beust.jcommander.Parameter;
 import com.beust.jcommander.Parameters;
 
-public class GenericEventAdapter
-{
+public class GenericEventAdapter {
 
-  public static void main(String[] args)
-  {
-    AdapterArgs adapterArgs = new AdapterArgs();
+    public static void main(String[] args) {
+        AdapterArgs adapterArgs = new AdapterArgs();
 
-    Tools.parseArgs(adapterArgs, args);
-    try
-    {
-      String instanceName = "adapter";
-      HelixManager manager = HelixManagerFactory.getZKHelixManager(
-          adapterArgs.clusterName, instanceName, InstanceType.SPECTATOR,
-          adapterArgs.zkConnectionString);
-      ClusterFromHelix cluster = new ClusterFromHelix(adapterArgs.clusterName,
-          adapterArgs.zkConnectionString, 30, 60);
-      manager.connect();
-      manager.addExternalViewChangeListener(cluster);
+        Tools.parseArgs(adapterArgs, args);
+        try {
+            String instanceName = "adapter";
+            HelixManager manager = HelixManagerFactory.getZKHelixManager(adapterArgs.clusterName, instanceName,
+                    InstanceType.SPECTATOR, adapterArgs.zkConnectionString);
+            ClusterFromHelix cluster = new ClusterFromHelix(adapterArgs.clusterName, adapterArgs.zkConnectionString,
+                    30, 60);
+            manager.connect();
+            manager.addExternalViewChangeListener(cluster);
 
-      HelixDataAccessor helixDataAccessor = manager.getHelixDataAccessor();
-      Builder keyBuilder = helixDataAccessor.keyBuilder();
-      IdealState idealstate = helixDataAccessor.getProperty(keyBuilder
-          .idealStates(adapterArgs.streamName));
-      TCPEmitter emitter = new TCPEmitter(cluster, 1000);
-      while (true)
-      {
-        int partitionId = ((int) (Math.random() * 1000))
-            % idealstate.getNumPartitions();
-        Event event = new Event();
-        event.put("name", String.class,
-            "Hello world to partition:" + partitionId);
-        KryoSerDeser serializer = new KryoSerDeser();
-        EventMessage message = new EventMessage("-1", adapterArgs.streamName,
-            serializer.serialize(event));
-        System.out.println("Sending event to partition:"+partitionId);
-        emitter.send(partitionId, message);
-        Thread.sleep(1000);
-      }
-    } catch (Exception e)
-    {
-      e.printStackTrace();
+            HelixDataAccessor helixDataAccessor = manager.getHelixDataAccessor();
+            Builder keyBuilder = helixDataAccessor.keyBuilder();
+            IdealState idealstate = helixDataAccessor.getProperty(keyBuilder.idealStates(adapterArgs.streamName));
+            TCPEmitter emitter = new TCPEmitter(cluster, 1000);
+            while (true) {
+                int partitionId = ((int) (Math.random() * 1000)) % idealstate.getNumPartitions();
+                Event event = new Event();
+                event.put("name", String.class, "Hello world to partition:" + partitionId);
+                KryoSerDeser serializer = new KryoSerDeser();
+                EventMessage message = new EventMessage("-1", adapterArgs.streamName, serializer.serialize(event));
+                System.out.println("Sending event to partition:" + partitionId);
+                emitter.send(partitionId, message);
+                Thread.sleep(1000);
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+
     }
 
-  }
+    @Parameters(commandNames = "newStreamProcessor", separators = "=", commandDescription = "Create a new stream processor")
+    static class AdapterArgs extends S4ArgsBase {
 
-  @Parameters(commandNames = "newStreamProcessor", separators = "=", commandDescription = "Create a new stream processor")
-  static class AdapterArgs extends S4ArgsBase
-  {
+        @Parameter(names = "-zk", description = "ZooKeeper connection string")
+        String zkConnectionString = "localhost:2181";
 
-    @Parameter(names = "-zk", description = "ZooKeeper connection string")
-    String zkConnectionString = "localhost:2181";
+        @Parameter(names = { "-c", "-cluster" }, description = "Logical name of the S4 cluster", required = true)
+        String clusterName;
 
-    @Parameter(names = { "-c", "-cluster" }, description = "Logical name of the S4 cluster", required = true)
-    String clusterName;
+        @Parameter(names = { "-s", "-streamName" }, description = "Stream Name where the event will be sent to", required = true)
+        String streamName;
 
-    @Parameter(names = { "-s", "-streamName" }, description = "Stream Name where the event will be sent to", required = true)
-    String streamName;
-
-  }
+    }
 
 }
diff --git a/subprojects/s4-tools/src/main/java/org/apache/s4/tools/RebalanceTask.java b/subprojects/s4-tools/src/main/java/org/apache/s4/tools/RebalanceTask.java
index 1b8214f..13b88ca 100644
--- a/subprojects/s4-tools/src/main/java/org/apache/s4/tools/RebalanceTask.java
+++ b/subprojects/s4-tools/src/main/java/org/apache/s4/tools/RebalanceTask.java
@@ -17,46 +17,43 @@
 import com.beust.jcommander.Parameter;
 import com.beust.jcommander.Parameters;
 
-public class RebalanceTask extends S4ArgsBase
-{
-  public static void main(String[] args)
-  {
-    RebalanceTaskArgs taskArgs = new RebalanceTaskArgs();
+public class RebalanceTask extends S4ArgsBase {
+    public static void main(String[] args) {
+        RebalanceTaskArgs taskArgs = new RebalanceTaskArgs();
 
-    Tools.parseArgs(taskArgs, args);
+        Tools.parseArgs(taskArgs, args);
 
-    HelixAdmin admin = new ZKHelixAdmin(taskArgs.zkConnectionString);
-    // This does the assignment of partition to nodes. It uses a modified
-    // version of consistent hashing to distribute active partitions and standbys
-    // equally among nodes.
-    IdealState currentAssignment = admin.getResourceIdealState(taskArgs.clusterName, taskArgs.taskId);
-    List<String> instancesInGroup = new ArrayList<String>();
-    List<String> instancesInCluster = admin.getInstancesInCluster(taskArgs.clusterName);
-    for (String instanceName : instancesInCluster) {
-        InstanceConfig instanceConfig = admin.getInstanceConfig(taskArgs.clusterName, instanceName);
-        String nodeGroup = instanceConfig.getRecord().getSimpleField("GROUP");
-        if (nodeGroup.equals(taskArgs.nodeGroup)) {
-            instancesInGroup.add(instanceName);
+        HelixAdmin admin = new ZKHelixAdmin(taskArgs.zkConnectionString);
+        // This does the assignment of partition to nodes. It uses a modified
+        // version of consistent hashing to distribute active partitions and standbys
+        // equally among nodes.
+        IdealState currentAssignment = admin.getResourceIdealState(taskArgs.clusterName, taskArgs.taskId);
+        List<String> instancesInGroup = new ArrayList<String>();
+        List<String> instancesInCluster = admin.getInstancesInCluster(taskArgs.clusterName);
+        for (String instanceName : instancesInCluster) {
+            InstanceConfig instanceConfig = admin.getInstanceConfig(taskArgs.clusterName, instanceName);
+            String nodeGroup = instanceConfig.getRecord().getSimpleField("GROUP");
+            if (nodeGroup.equals(taskArgs.nodeGroup)) {
+                instancesInGroup.add(instanceName);
+            }
         }
+        admin.rebalance(taskArgs.clusterName, currentAssignment, instancesInGroup);
     }
-    admin.rebalance(taskArgs.clusterName,currentAssignment, instancesInGroup);
-  }
 
-  @Parameters(commandNames = "newStreamProcessor", separators = "=", commandDescription = "Create a new stream processor")
-  static class RebalanceTaskArgs extends S4ArgsBase
-  {
+    @Parameters(commandNames = "newStreamProcessor", separators = "=", commandDescription = "Create a new stream processor")
+    static class RebalanceTaskArgs extends S4ArgsBase {
 
-    @Parameter(names = "-zk", description = "ZooKeeper connection string")
-    String zkConnectionString = "localhost:2181";
+        @Parameter(names = "-zk", description = "ZooKeeper connection string")
+        String zkConnectionString = "localhost:2181";
 
-    @Parameter(names = { "-c", "-cluster" }, description = "Logical name of the S4 cluster", required = true)
-    String clusterName;
+        @Parameter(names = { "-c", "-cluster" }, description = "Logical name of the S4 cluster", required = true)
+        String clusterName;
 
-    @Parameter(names={"-id","-taskId"},description = "id of the task that produces/consumes a stream", required = true, arity = 1)
-    String taskId;
+        @Parameter(names = { "-id", "-taskId" }, description = "id of the task that produces/consumes a stream", required = true, arity = 1)
+        String taskId;
 
-    @Parameter(names = { "-ng", "-nodeGroup" }, description = "Node group name where the task needs to be run", required = true, arity = 1)
-    String nodeGroup = "default";
+        @Parameter(names = { "-ng", "-nodeGroup" }, description = "Node group name where the task needs to be run", required = true, arity = 1)
+        String nodeGroup = "default";
 
-  }
+    }
 }
diff --git a/subprojects/s4-tools/src/main/java/org/apache/s4/tools/S4Status.java b/subprojects/s4-tools/src/main/java/org/apache/s4/tools/S4Status.java
index bd4224e..2be58c1 100644
--- a/subprojects/s4-tools/src/main/java/org/apache/s4/tools/S4Status.java
+++ b/subprojects/s4-tools/src/main/java/org/apache/s4/tools/S4Status.java
@@ -54,521 +54,413 @@
 import com.beust.jcommander.Parameters;
 import com.google.common.collect.Maps;
 
-public class S4Status extends S4ArgsBase
-{
-  static Logger logger = LoggerFactory.getLogger(S4Status.class);
+public class S4Status extends S4ArgsBase {
+    static Logger logger = LoggerFactory.getLogger(S4Status.class);
 
-  private static String NONE = "--";
+    private static String NONE = "--";
 
-  public static void main(String[] args)
-  {
+    public static void main(String[] args) {
 
-    StatusArgs statusArgs = new StatusArgs();
-    Tools.parseArgs(statusArgs, args);
+        StatusArgs statusArgs = new StatusArgs();
+        Tools.parseArgs(statusArgs, args);
 
-    try
-    {
-      if (statusArgs.clusters.size() > 0)
-      {
-        for (String cluster : statusArgs.clusters)
-        {
-          HelixManager manager = HelixManagerFactory.getZKHelixManager(cluster,
-              "ADMIN", InstanceType.ADMINISTRATOR,
-              statusArgs.zkConnectionString);
-          manager.connect();
-          ConfigAccessor configAccessor = manager.getConfigAccessor();
-          ConfigScopeBuilder builder = new ConfigScopeBuilder();
-          printClusterInfo(manager, cluster);
-          List<String> resourcesInCluster = manager.getClusterManagmentTool()
-              .getResourcesInCluster(cluster);
+        try {
+            if (statusArgs.clusters.size() > 0) {
+                for (String cluster : statusArgs.clusters) {
+                    HelixManager manager = HelixManagerFactory.getZKHelixManager(cluster, "ADMIN",
+                            InstanceType.ADMINISTRATOR, statusArgs.zkConnectionString);
+                    manager.connect();
+                    ConfigAccessor configAccessor = manager.getConfigAccessor();
+                    ConfigScopeBuilder builder = new ConfigScopeBuilder();
+                    printClusterInfo(manager, cluster);
+                    List<String> resourcesInCluster = manager.getClusterManagmentTool().getResourcesInCluster(cluster);
 
-          List<String> apps = new ArrayList<String>();
-          List<String> tasks = new ArrayList<String>();
+                    List<String> apps = new ArrayList<String>();
+                    List<String> tasks = new ArrayList<String>();
 
-          for (String resource : resourcesInCluster)
-          {
-            ConfigScope scope = builder.forCluster(cluster)
-                .forResource(resource).build();
-            String resourceType = configAccessor.get(scope, "type");
-            if ("App".equals(resourceType))
-            {
-              apps.add(resource);
-            } else if ("Task".equals(resourceType))
-            {
-              tasks.add(resource);
+                    for (String resource : resourcesInCluster) {
+                        ConfigScope scope = builder.forCluster(cluster).forResource(resource).build();
+                        String resourceType = configAccessor.get(scope, "type");
+                        if ("App".equals(resourceType)) {
+                            apps.add(resource);
+                        } else if ("Task".equals(resourceType)) {
+                            tasks.add(resource);
+                        }
+                    }
+                    if (statusArgs.apps == null && statusArgs.streams == null) {
+                        statusArgs.apps = apps;
+                        statusArgs.streams = tasks;
+                    }
+                    for (String app : statusArgs.apps) {
+                        if (resourcesInCluster.contains(app)) {
+                            printAppInfo(manager, cluster, app);
+                        }
+                    }
+
+                    if (statusArgs.streams != null && statusArgs.streams.size() > 0) {
+                        for (String stream : statusArgs.streams) {
+                            if (resourcesInCluster.contains(stream)) {
+                                printStreamInfo(manager, cluster, stream);
+                            }
+                        }
+                    }
+                    manager.disconnect();
+                }
             }
-          }
-          if (statusArgs.apps == null && statusArgs.streams == null)
-          {
-            statusArgs.apps = apps;
-            statusArgs.streams = tasks;
-          }
-          for (String app : statusArgs.apps)
-          {
-            if (resourcesInCluster.contains(app))
-            {
-              printAppInfo(manager, cluster, app);
+        } catch (Exception e) {
+            logger.error("Cannot get the status of S4", e);
+        }
+
+    }
+
+    private static void printStreamInfo(HelixManager manager, String cluster, String taskId) {
+        ConfigAccessor configAccessor = manager.getConfigAccessor();
+        ConfigScopeBuilder builder = new ConfigScopeBuilder();
+        ConfigScope scope = builder.forCluster(cluster).forResource(taskId).build();
+        String streamName = configAccessor.get(scope, "streamName");
+        String taskType = configAccessor.get(scope, "taskType");
+
+        System.out.println("Task Status");
+        System.out.println(generateEdge(130));
+        System.out.format("%-20s%-20s%-90s%n", inMiddle("Task Id", 20), inMiddle("Cluster", 20),
+                inMiddle("Description", 90));
+        System.out.println(generateEdge(130));
+        System.out.format("%-20s%-20s%-90s%n", inMiddle(taskId, 20), inMiddle(cluster, 20),
+                inMiddle(streamName + " " + taskType, 90));
+        System.out.println(generateEdge(130));
+        HelixDataAccessor helixDataAccessor = manager.getHelixDataAccessor();
+        Builder keyBuilder = helixDataAccessor.keyBuilder();
+        IdealState assignment = helixDataAccessor.getProperty(keyBuilder.idealStates(taskId));
+        ExternalView view = helixDataAccessor.getProperty(keyBuilder.externalView(taskId));
+        List<String> liveInstances = helixDataAccessor.getChildNames(keyBuilder.liveInstances());
+        System.out.format("%-50s%-100s%n", inMiddle("Partition", 50), inMiddle("State", 20));
+        System.out.println(generateEdge(130));
+        for (String partition : assignment.getPartitionSet()) {
+            Map<String, String> stateMap = view.getStateMap(partition);
+            StringBuilder sb = new StringBuilder();
+            String delim = "";
+            for (String instance : stateMap.keySet()) {
+                sb.append(delim);
+                String state = stateMap.get(instance);
+                if (liveInstances.contains(instance)) {
+                    sb.append(instance).append(":").append(state);
+                } else {
+                    sb.append(instance).append(":").append("OFFLINE");
+                }
+                delim = ", ";
             }
-          }
+            System.out.format("%-50s%-10s%n", inMiddle(partition, 50), inMiddle(sb.toString(), 100));
+        }
+        System.out.println(generateEdge(130));
+    }
 
-          if (statusArgs.streams != null && statusArgs.streams.size() > 0)
-          {
-            for (String stream : statusArgs.streams)
-            {
-              if (resourcesInCluster.contains(stream))
-              {
-                printStreamInfo(manager, cluster, stream);
-              }
+    private static void printAppInfo(HelixManager manager, String cluster, String app) {
+        ConfigAccessor configAccessor = manager.getConfigAccessor();
+        ConfigScopeBuilder builder = new ConfigScopeBuilder();
+        ConfigScope scope = builder.forCluster(cluster).forResource(app).build();
+        String uri = configAccessor.get(scope, "s4r_uri");
+
+        System.out.println("App Status");
+        System.out.println(generateEdge(130));
+        System.out.format("%-20s%-20s%-90s%n", inMiddle("Name", 20), inMiddle("Cluster", 20), inMiddle("URI", 90));
+        System.out.println(generateEdge(130));
+        System.out.format("%-20s%-20s%-90s%n", inMiddle(app, 20), inMiddle(cluster, 20), inMiddle(uri, 90));
+        System.out.println(generateEdge(130));
+        HelixDataAccessor helixDataAccessor = manager.getHelixDataAccessor();
+        Builder keyBuilder = helixDataAccessor.keyBuilder();
+        IdealState assignment = helixDataAccessor.getProperty(keyBuilder.idealStates(app));
+        ExternalView view = helixDataAccessor.getProperty(keyBuilder.externalView(app));
+        List<String> liveInstances = helixDataAccessor.getChildNames(keyBuilder.liveInstances());
+        Map<String, String> assignmentMap = assignment.getInstanceStateMap(app);
+        Map<String, String> appStateMap = view.getStateMap(app);
+        System.out.format("%-50s%-20s%n", inMiddle("Node id", 50), inMiddle("DEPLOYED", 20));
+        System.out.println(generateEdge(130));
+        for (String instance : assignmentMap.keySet()) {
+            String state = appStateMap.get(instance);
+            System.out.format("%-50s%-10s%n", inMiddle(instance, 50),
+                    inMiddle((("ONLINE".equals(state) && liveInstances.contains(instance)) ? "Y" : "N"), 20));
+        }
+
+        System.out.println(generateEdge(130));
+
+    }
+
+    private static void printClusterInfo(HelixManager manager, String cluster) {
+        HelixDataAccessor dataAccessor = manager.getHelixDataAccessor();
+        Builder keyBuilder = dataAccessor.keyBuilder();
+        List<String> instances = dataAccessor.getChildNames(keyBuilder.instanceConfigs());
+        List<String> liveInstances = dataAccessor.getChildNames(keyBuilder.liveInstances());
+        if (liveInstances == null) {
+            liveInstances = Collections.emptyList();
+        }
+        System.out.println("Cluster Status");
+        System.out.println(generateEdge(130));
+        System.out.format("%-50s%-80s%n", " ", inMiddle("Nodes", 80));
+        System.out.format("%-20s%-20s%-10s%s%n", inMiddle("Cluster Name", 20), inMiddle("Nodes", 20),
+                inMiddle("Active", 10), generateEdge(80));
+        System.out.format("%-54s%-10s%-50s%-8s%-8s%n", " ", inMiddle("Node id", 10), inMiddle("Host", 50),
+                inMiddle("Port", 8), inMiddle("Active", 10));
+        System.out.println(generateEdge(130));
+
+        System.out.format("%-20s%-20s%-10s", inMiddle(cluster, 20), inMiddle("" + instances.size(), 8),
+                inMiddle("" + liveInstances.size(), 8));
+        boolean first = true;
+
+        for (String instance : instances) {
+            InstanceConfig config = dataAccessor.getProperty(keyBuilder.instanceConfig(instance));
+            // System.out.println(config);
+            if (first) {
+                first = false;
+            } else {
+                System.out.format("%n%-50s", " ");
             }
-          }
-          manager.disconnect();
+            System.out.format("%-10s%-46s%-10s%-10s", inMiddle("" + config.getId(), 10),
+                    inMiddle(config.getHostName(), 50), inMiddle(config.getPort() + "", 10),
+                    inMiddle(liveInstances.contains(config.getInstanceName()) ? "Y" : "N", 10));
         }
-      }
-    } catch (Exception e)
-    {
-      logger.error("Cannot get the status of S4", e);
+
+        System.out.println();
     }
 
-  }
+    @Parameters(commandNames = "s4 status", commandDescription = "Show status of S4", separators = "=")
+    static class StatusArgs extends S4ArgsBase {
 
-  private static void printStreamInfo(HelixManager manager, String cluster,
-      String taskId)
-  {
-    ConfigAccessor configAccessor = manager.getConfigAccessor();
-    ConfigScopeBuilder builder = new ConfigScopeBuilder();
-    ConfigScope scope = builder.forCluster(cluster).forResource(taskId).build();
-    String streamName = configAccessor.get(scope, "streamName");
-    String taskType = configAccessor.get(scope, "taskType");
+        @Parameter(names = { "-app" }, description = "Only show status of specified S4 application(s)", required = false)
+        List<String> apps;
 
-    System.out.println("Task Status");
-    System.out.println(generateEdge(130));
-    System.out.format("%-20s%-20s%-90s%n", inMiddle("Task Id", 20),
-        inMiddle("Cluster", 20), inMiddle("Description", 90));
-    System.out.println(generateEdge(130));
-    System.out.format("%-20s%-20s%-90s%n", inMiddle(taskId, 20),
-        inMiddle(cluster, 20), inMiddle(streamName + " " + taskType, 90));
-    System.out.println(generateEdge(130));
-    HelixDataAccessor helixDataAccessor = manager.getHelixDataAccessor();
-    Builder keyBuilder = helixDataAccessor.keyBuilder();
-    IdealState assignment = helixDataAccessor.getProperty(keyBuilder
-        .idealStates(taskId));
-    ExternalView view = helixDataAccessor.getProperty(keyBuilder
-        .externalView(taskId));
-    List<String> liveInstances = helixDataAccessor.getChildNames(keyBuilder
-        .liveInstances());
-    System.out.format("%-50s%-100s%n", inMiddle("Partition", 50),
-        inMiddle("State", 20));
-    System.out.println(generateEdge(130));
-    for (String partition : assignment.getPartitionSet())
-    {
-      Map<String, String> stateMap = view.getStateMap(partition);
-      StringBuilder sb = new StringBuilder();
-      String delim="";
-      for (String instance : stateMap.keySet())
-      {
-        sb.append(delim);
-        String state = stateMap.get(instance);
-        if (liveInstances.contains(instance))
-        {
-          sb.append(instance).append(":").append(state);
-        } else
-        {
-          sb.append(instance).append(":").append("OFFLINE");
+        @Parameter(names = { "-c", "-cluster" }, description = "Only show status of specified S4 cluster(s)", required = true)
+        List<String> clusters;
+
+        @Parameter(names = { "-s", "-stream" }, description = "Only show status of specified published stream(s)", required = false)
+        List<String> streams;
+
+        @Parameter(names = "-zk", description = "ZooKeeper connection string")
+        String zkConnectionString = "localhost:2181";
+
+        @Parameter(names = "-timeout", description = "Connection timeout to Zookeeper, in ms")
+        int timeout = 10000;
+    }
+
+    private static void showAppsStatus(List<Cluster> clusters) {
+        System.out.println("App Status");
+        System.out.println(generateEdge(130));
+        System.out.format("%-20s%-20s%-90s%n", inMiddle("Name", 20), inMiddle("Cluster", 20), inMiddle("URI", 90));
+        System.out.println(generateEdge(130));
+        for (Cluster cluster : clusters) {
+            if (!NONE.equals(cluster.app.name)) {
+                System.out.format("%-20s%-20s%-90s%n", inMiddle(cluster.app.name, 20),
+                        inMiddle(cluster.app.cluster, 20), cluster.app.uri);
+            }
         }
-       delim=", ";
-      }
-      System.out.format("%-50s%-10s%n", inMiddle(partition, 50),
-          inMiddle(sb.toString(), 100));
-    }
-    System.out.println(generateEdge(130));
-  }
+        System.out.println(generateEdge(130));
 
-  private static void printAppInfo(HelixManager manager, String cluster,
-      String app)
-  {
-    ConfigAccessor configAccessor = manager.getConfigAccessor();
-    ConfigScopeBuilder builder = new ConfigScopeBuilder();
-    ConfigScope scope = builder.forCluster(cluster).forResource(app).build();
-    String uri = configAccessor.get(scope, "s4r_uri");
-
-    System.out.println("App Status");
-    System.out.println(generateEdge(130));
-    System.out.format("%-20s%-20s%-90s%n", inMiddle("Name", 20),
-        inMiddle("Cluster", 20), inMiddle("URI", 90));
-    System.out.println(generateEdge(130));
-    System.out.format("%-20s%-20s%-90s%n", inMiddle(app, 20),
-        inMiddle(cluster, 20), inMiddle(uri, 90));
-    System.out.println(generateEdge(130));
-    HelixDataAccessor helixDataAccessor = manager.getHelixDataAccessor();
-    Builder keyBuilder = helixDataAccessor.keyBuilder();
-    IdealState assignment = helixDataAccessor.getProperty(keyBuilder
-        .idealStates(app));
-    ExternalView view = helixDataAccessor.getProperty(keyBuilder
-        .externalView(app));
-    List<String> liveInstances = helixDataAccessor.getChildNames(keyBuilder
-        .liveInstances());
-    Map<String, String> assignmentMap = assignment.getInstanceStateMap(app);
-    Map<String, String> appStateMap = view.getStateMap(app);
-    System.out.format("%-50s%-20s%n", inMiddle("Node id", 50),
-        inMiddle("DEPLOYED", 20));
-    System.out.println(generateEdge(130));
-    for (String instance : assignmentMap.keySet())
-    {
-      String state = appStateMap.get(instance);
-      System.out
-          .format(
-              "%-50s%-10s%n",
-              inMiddle(instance, 50),
-              inMiddle((("ONLINE".equals(state) && liveInstances
-                  .contains(instance)) ? "Y" : "N"), 20));
     }
 
-    System.out.println(generateEdge(130));
+    private static void showClustersStatus(List<Cluster> clusters) {
+        System.out.println("Cluster Status");
+        System.out.println(generateEdge(130));
+        System.out.format("%-50s%-80s%n", " ", inMiddle("Active nodes", 80));
+        System.out.format("%-20s%-20s%-10s%s%n", inMiddle("Name", 20), inMiddle("App", 20), inMiddle("Tasks", 10),
+                generateEdge(80));
+        System.out.format("%-50s%-10s%-10s%-50s%-10s%n", " ", inMiddle("Number", 8), inMiddle("Task id", 10),
+                inMiddle("Host", 50), inMiddle("Port", 8));
+        System.out.println(generateEdge(130));
 
-  }
-
-  private static void printClusterInfo(HelixManager manager, String cluster)
-  {
-    HelixDataAccessor dataAccessor = manager.getHelixDataAccessor();
-    Builder keyBuilder = dataAccessor.keyBuilder();
-    List<String> instances = dataAccessor.getChildNames(keyBuilder
-        .instanceConfigs());
-    List<String> liveInstances = dataAccessor.getChildNames(keyBuilder
-        .liveInstances());
-    if (liveInstances == null)
-    {
-      liveInstances = Collections.emptyList();
-    }
-    System.out.println("Cluster Status");
-    System.out.println(generateEdge(130));
-    System.out.format("%-50s%-80s%n", " ", inMiddle("Nodes", 80));
-    System.out.format("%-20s%-20s%-10s%s%n", inMiddle("Cluster Name", 20),
-        inMiddle("Nodes", 20), inMiddle("Active", 10), generateEdge(80));
-    System.out.format("%-54s%-10s%-50s%-8s%-8s%n", " ",
-        inMiddle("Node id", 10), inMiddle("Host", 50), inMiddle("Port", 8),
-        inMiddle("Active", 10));
-    System.out.println(generateEdge(130));
-
-    System.out.format("%-20s%-20s%-10s", inMiddle(cluster, 20),
-        inMiddle("" + instances.size(), 8),
-        inMiddle("" + liveInstances.size(), 8));
-    boolean first = true;
-
-    for (String instance : instances)
-    {
-      InstanceConfig config = dataAccessor.getProperty(keyBuilder
-          .instanceConfig(instance));
-      // System.out.println(config);
-      if (first)
-      {
-        first = false;
-      } else
-      {
-        System.out.format("%n%-50s", " ");
-      }
-      System.out
-          .format(
-              "%-10s%-46s%-10s%-10s",
-              inMiddle("" + config.getId(), 10),
-              inMiddle(config.getHostName(), 50),
-              inMiddle(config.getPort() + "", 10),
-              inMiddle(liveInstances.contains(config.getInstanceName()) ? "Y"
-                  : "N", 10));
-    }
-
-    System.out.println();
-  }
-
-  @Parameters(commandNames = "s4 status", commandDescription = "Show status of S4", separators = "=")
-  static class StatusArgs extends S4ArgsBase
-  {
-
-    @Parameter(names = { "-app" }, description = "Only show status of specified S4 application(s)", required = false)
-    List<String> apps;
-
-    @Parameter(names = { "-c", "-cluster" }, description = "Only show status of specified S4 cluster(s)", required = true)
-    List<String> clusters;
-
-    @Parameter(names = { "-s", "-stream" }, description = "Only show status of specified published stream(s)", required = false)
-    List<String> streams;
-
-    @Parameter(names = "-zk", description = "ZooKeeper connection string")
-    String zkConnectionString = "localhost:2181";
-
-    @Parameter(names = "-timeout", description = "Connection timeout to Zookeeper, in ms")
-    int timeout = 10000;
-  }
-
-  private static void showAppsStatus(List<Cluster> clusters)
-  {
-    System.out.println("App Status");
-    System.out.println(generateEdge(130));
-    System.out.format("%-20s%-20s%-90s%n", inMiddle("Name", 20),
-        inMiddle("Cluster", 20), inMiddle("URI", 90));
-    System.out.println(generateEdge(130));
-    for (Cluster cluster : clusters)
-    {
-      if (!NONE.equals(cluster.app.name))
-      {
-        System.out.format("%-20s%-20s%-90s%n", inMiddle(cluster.app.name, 20),
-            inMiddle(cluster.app.cluster, 20), cluster.app.uri);
-      }
-    }
-    System.out.println(generateEdge(130));
-
-  }
-
-  private static void showClustersStatus(List<Cluster> clusters)
-  {
-    System.out.println("Cluster Status");
-    System.out.println(generateEdge(130));
-    System.out.format("%-50s%-80s%n", " ", inMiddle("Active nodes", 80));
-    System.out.format("%-20s%-20s%-10s%s%n", inMiddle("Name", 20),
-        inMiddle("App", 20), inMiddle("Tasks", 10), generateEdge(80));
-    System.out.format("%-50s%-10s%-10s%-50s%-10s%n", " ",
-        inMiddle("Number", 8), inMiddle("Task id", 10), inMiddle("Host", 50),
-        inMiddle("Port", 8));
-    System.out.println(generateEdge(130));
-
-    for (Cluster cluster : clusters)
-    {
-      System.out.format("%-20s%-20s%-10s%-10s",
-          inMiddle(cluster.clusterName, 20), inMiddle(cluster.app.name, 20),
-          inMiddle("" + cluster.taskNumber, 8),
-          inMiddle("" + cluster.nodes.size(), 8));
-      boolean first = true;
-      for (ClusterNode node : cluster.nodes)
-      {
-        if (first)
-        {
-          first = false;
-        } else
-        {
-          System.out.format("%n%-60s", " ");
+        for (Cluster cluster : clusters) {
+            System.out.format("%-20s%-20s%-10s%-10s", inMiddle(cluster.clusterName, 20),
+                    inMiddle(cluster.app.name, 20), inMiddle("" + cluster.taskNumber, 8),
+                    inMiddle("" + cluster.nodes.size(), 8));
+            boolean first = true;
+            for (ClusterNode node : cluster.nodes) {
+                if (first) {
+                    first = false;
+                } else {
+                    System.out.format("%n%-60s", " ");
+                }
+                System.out.format("%-10s%-50s%-10s", inMiddle("" + node.getTaskId(), 10),
+                        inMiddle(node.getMachineName(), 50), inMiddle(node.getPort() + "", 10));
+            }
+            System.out.println();
         }
-        System.out.format("%-10s%-50s%-10s",
-            inMiddle("" + node.getTaskId(), 10),
-            inMiddle(node.getMachineName(), 50),
-            inMiddle(node.getPort() + "", 10));
-      }
-      System.out.println();
+        System.out.println(generateEdge(130));
     }
-    System.out.println(generateEdge(130));
-  }
 
-  private static void showStreamsStatus(List<Stream> streams)
-  {
-    System.out.println("Stream Status");
-    System.out.println(generateEdge(130));
-    System.out.format("%-20s%-55s%-55s%n", inMiddle("Name", 20),
-        inMiddle("Producers", 55), inMiddle("Consumers", 55));
-    System.out.println(generateEdge(130));
+    private static void showStreamsStatus(List<Stream> streams) {
+        System.out.println("Stream Status");
+        System.out.println(generateEdge(130));
+        System.out.format("%-20s%-55s%-55s%n", inMiddle("Name", 20), inMiddle("Producers", 55),
+                inMiddle("Consumers", 55));
+        System.out.println(generateEdge(130));
 
-    for (Stream stream : streams)
-    {
-      System.out
-          .format(
-              "%-20s%-55s%-55s%n",
-              inMiddle(stream.streamName, 20),
-              inMiddle(getFormatString(stream.producers, stream.clusterAppMap),
-                  55),
-              inMiddle(getFormatString(stream.consumers, stream.clusterAppMap),
-                  55));
-    }
-    System.out.println(generateEdge(130));
-
-  }
-
-  private static String inMiddle(String content, int width)
-  {
-    int i = (width - content.length()) / 2;
-    return String.format("%" + i + "s%s", " ", content);
-  }
-
-  private static String generateEdge(int length)
-  {
-    StringBuilder sb = new StringBuilder();
-    for (int i = 0; i < length; i++)
-    {
-      sb.append("-");
-    }
-    return sb.toString();
-  }
-
-  /**
-   * show as cluster1(app1), cluster2(app2)
-   * 
-   * @param clusters
-   *          cluster list
-   * @param clusterAppMap
-   *          <cluster,app>
-   * @return
-   */
-  private static String getFormatString(Collection<String> clusters,
-      Map<String, String> clusterAppMap)
-  {
-    if (clusters == null || clusters.size() == 0)
-    {
-      return NONE;
-    } else
-    {
-      // show as: cluster1(app1), cluster2(app2)
-      StringBuilder sb = new StringBuilder();
-      for (String cluster : clusters)
-      {
-        String app = clusterAppMap.get(cluster);
-        sb.append(cluster);
-        if (!NONE.equals(app))
-        {
-          sb.append("(").append(app).append(")");
+        for (Stream stream : streams) {
+            System.out.format("%-20s%-55s%-55s%n", inMiddle(stream.streamName, 20),
+                    inMiddle(getFormatString(stream.producers, stream.clusterAppMap), 55),
+                    inMiddle(getFormatString(stream.consumers, stream.clusterAppMap), 55));
         }
-        sb.append(" ");
-      }
-      return sb.toString();
-    }
-  }
+        System.out.println(generateEdge(130));
 
-  static class Stream
-  {
-
-    private final ZkClient zkClient;
-    private final String consumerPath;
-    private final String producerPath;
-
-    String streamName;
-    Set<String> producers = new HashSet<String>();// cluster name
-    Set<String> consumers = new HashSet<String>();// cluster name
-
-    Map<String, String> clusterAppMap = Maps.newHashMap();
-
-    public Stream(String streamName, ZkClient zkClient) throws Exception
-    {
-      this.streamName = streamName;
-      this.zkClient = zkClient;
-      this.consumerPath = "/s4/streams/" + streamName + "/consumers";
-      this.producerPath = "/s4/streams/" + streamName + "/producers";
-      readStreamFromZk();
     }
 
-    private void readStreamFromZk() throws Exception
-    {
-      List<String> consumerNodes = zkClient.getChildren(consumerPath);
-      for (String node : consumerNodes)
-      {
-        ZNRecord consumer = zkClient.readData(consumerPath + "/" + node, true);
-        consumers.add(consumer.getSimpleField("clusterName"));
-      }
-
-      List<String> producerNodes = zkClient.getChildren(producerPath);
-      for (String node : producerNodes)
-      {
-        ZNRecord consumer = zkClient.readData(producerPath + "/" + node, true);
-        producers.add(consumer.getSimpleField("clusterName"));
-      }
-
-      getAppNames();
+    private static String inMiddle(String content, int width) {
+        int i = (width - content.length()) / 2;
+        return String.format("%" + i + "s%s", " ", content);
     }
 
-    private void getAppNames()
-    {
-      Set<String> clusters = new HashSet<String>(consumers);
-      clusters.addAll(producers);
-      for (String cluster : clusters)
-      {
-        clusterAppMap.put(cluster, getApp(cluster, zkClient));
-      }
-    }
-
-    public boolean containsCluster(String cluster)
-    {
-      if (producers.contains(cluster) || consumers.contains(cluster))
-      {
-        return true;
-      }
-      return false;
-    }
-
-    private static String getApp(String clusterName, ZkClient zkClient)
-    {
-      String appPath = "/s4/clusters/" + clusterName + "/app/s4App";
-      if (zkClient.exists(appPath))
-      {
-        ZNRecord appRecord = zkClient.readData("/s4/clusters/" + clusterName
-            + "/app/s4App");
-        return appRecord.getSimpleField("name");
-      }
-      return NONE;
-    }
-  }
-
-  static class App
-  {
-    private String name = NONE;
-    private String cluster;
-    private String uri = NONE;
-  }
-
-  static class Cluster
-  {
-    private final ZkClient zkClient;
-    private final String taskPath;
-    private final String processPath;
-    private final String appPath;
-
-    String clusterName;
-    int taskNumber;
-    App app;
-
-    List<ClusterNode> nodes = new ArrayList<ClusterNode>();
-
-    public Cluster(String clusterName, ZkClient zkClient) throws Exception
-    {
-      this.clusterName = clusterName;
-      this.zkClient = zkClient;
-      this.taskPath = "/s4/clusters/" + clusterName + "/tasks";
-      this.processPath = "/s4/clusters/" + clusterName + "/process";
-      this.appPath = "/s4/clusters/" + clusterName + "/app/s4App";
-      readClusterFromZk();
-    }
-
-    public void readClusterFromZk() throws Exception
-    {
-      List<String> processes;
-      List<String> tasks;
-
-      tasks = zkClient.getChildren(taskPath);
-      processes = zkClient.getChildren(processPath);
-
-      taskNumber = tasks.size();
-
-      for (int i = 0; i < processes.size(); i++)
-      {
-        ZNRecord process = zkClient.readData(
-            processPath + "/" + processes.get(i), true);
-        if (process != null)
-        {
-          int partition = Integer.parseInt(process.getSimpleField("partition"));
-          String host = process.getSimpleField("host");
-          int port = Integer.parseInt(process.getSimpleField("port"));
-          String taskId = process.getSimpleField("taskId");
-          ClusterNode node = new ClusterNode(partition, port, host, taskId);
-          nodes.add(node);
+    private static String generateEdge(int length) {
+        StringBuilder sb = new StringBuilder();
+        for (int i = 0; i < length; i++) {
+            sb.append("-");
         }
-      }
-
-      app = new App();
-      app.cluster = clusterName;
-      try
-      {
-        ZNRecord appRecord = zkClient.readData(appPath);
-        app.name = appRecord.getSimpleField("name");
-        app.uri = appRecord.getSimpleField("s4r_uri");
-      } catch (ZkNoNodeException e)
-      {
-        logger.warn(appPath + " doesn't exist");
-      }
+        return sb.toString();
     }
 
-  }
+    /**
+     * show as cluster1(app1), cluster2(app2)
+     * 
+     * @param clusters
+     *            cluster list
+     * @param clusterAppMap
+     *            <cluster,app>
+     * @return
+     */
+    private static String getFormatString(Collection<String> clusters, Map<String, String> clusterAppMap) {
+        if (clusters == null || clusters.size() == 0) {
+            return NONE;
+        } else {
+            // show as: cluster1(app1), cluster2(app2)
+            StringBuilder sb = new StringBuilder();
+            for (String cluster : clusters) {
+                String app = clusterAppMap.get(cluster);
+                sb.append(cluster);
+                if (!NONE.equals(app)) {
+                    sb.append("(").append(app).append(")");
+                }
+                sb.append(" ");
+            }
+            return sb.toString();
+        }
+    }
+
+    static class Stream {
+
+        private final ZkClient zkClient;
+        private final String consumerPath;
+        private final String producerPath;
+
+        String streamName;
+        Set<String> producers = new HashSet<String>();// cluster name
+        Set<String> consumers = new HashSet<String>();// cluster name
+
+        Map<String, String> clusterAppMap = Maps.newHashMap();
+
+        public Stream(String streamName, ZkClient zkClient) throws Exception {
+            this.streamName = streamName;
+            this.zkClient = zkClient;
+            this.consumerPath = "/s4/streams/" + streamName + "/consumers";
+            this.producerPath = "/s4/streams/" + streamName + "/producers";
+            readStreamFromZk();
+        }
+
+        private void readStreamFromZk() throws Exception {
+            List<String> consumerNodes = zkClient.getChildren(consumerPath);
+            for (String node : consumerNodes) {
+                ZNRecord consumer = zkClient.readData(consumerPath + "/" + node, true);
+                consumers.add(consumer.getSimpleField("clusterName"));
+            }
+
+            List<String> producerNodes = zkClient.getChildren(producerPath);
+            for (String node : producerNodes) {
+                ZNRecord consumer = zkClient.readData(producerPath + "/" + node, true);
+                producers.add(consumer.getSimpleField("clusterName"));
+            }
+
+            getAppNames();
+        }
+
+        private void getAppNames() {
+            Set<String> clusters = new HashSet<String>(consumers);
+            clusters.addAll(producers);
+            for (String cluster : clusters) {
+                clusterAppMap.put(cluster, getApp(cluster, zkClient));
+            }
+        }
+
+        public boolean containsCluster(String cluster) {
+            if (producers.contains(cluster) || consumers.contains(cluster)) {
+                return true;
+            }
+            return false;
+        }
+
+        private static String getApp(String clusterName, ZkClient zkClient) {
+            String appPath = "/s4/clusters/" + clusterName + "/app/s4App";
+            if (zkClient.exists(appPath)) {
+                ZNRecord appRecord = zkClient.readData("/s4/clusters/" + clusterName + "/app/s4App");
+                return appRecord.getSimpleField("name");
+            }
+            return NONE;
+        }
+    }
+
+    static class App {
+        private String name = NONE;
+        private String cluster;
+        private String uri = NONE;
+    }
+
+    static class Cluster {
+        private final ZkClient zkClient;
+        private final String taskPath;
+        private final String processPath;
+        private final String appPath;
+
+        String clusterName;
+        int taskNumber;
+        App app;
+
+        List<ClusterNode> nodes = new ArrayList<ClusterNode>();
+
+        public Cluster(String clusterName, ZkClient zkClient) throws Exception {
+            this.clusterName = clusterName;
+            this.zkClient = zkClient;
+            this.taskPath = "/s4/clusters/" + clusterName + "/tasks";
+            this.processPath = "/s4/clusters/" + clusterName + "/process";
+            this.appPath = "/s4/clusters/" + clusterName + "/app/s4App";
+            readClusterFromZk();
+        }
+
+        public void readClusterFromZk() throws Exception {
+            List<String> processes;
+            List<String> tasks;
+
+            tasks = zkClient.getChildren(taskPath);
+            processes = zkClient.getChildren(processPath);
+
+            taskNumber = tasks.size();
+
+            for (int i = 0; i < processes.size(); i++) {
+                ZNRecord process = zkClient.readData(processPath + "/" + processes.get(i), true);
+                if (process != null) {
+                    int partition = Integer.parseInt(process.getSimpleField("partition"));
+                    String host = process.getSimpleField("host");
+                    int port = Integer.parseInt(process.getSimpleField("port"));
+                    String taskId = process.getSimpleField("taskId");
+                    ClusterNode node = new ClusterNode(partition, port, host, taskId);
+                    nodes.add(node);
+                }
+            }
+
+            app = new App();
+            app.cluster = clusterName;
+            try {
+                ZNRecord appRecord = zkClient.readData(appPath);
+                app.name = appRecord.getSimpleField("name");
+                app.uri = appRecord.getSimpleField("s4r_uri");
+            } catch (ZkNoNodeException e) {
+                logger.warn(appPath + " doesn't exist");
+            }
+        }
+
+    }
 
 }
diff --git a/subprojects/s4-tools/src/main/java/org/apache/s4/tools/Status.java b/subprojects/s4-tools/src/main/java/org/apache/s4/tools/Status.java
index 824aa03..ab0ba3e 100644
--- a/subprojects/s4-tools/src/main/java/org/apache/s4/tools/Status.java
+++ b/subprojects/s4-tools/src/main/java/org/apache/s4/tools/Status.java
@@ -219,7 +219,7 @@
 
     /**
      * show as cluster1(app1), cluster2(app2)
-     *
+     * 
      * @param clusters
      *            cluster list
      * @param clusterAppMap
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 192869a..3940c6d 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
@@ -38,20 +38,11 @@
 
     static Logger logger = LoggerFactory.getLogger(Tools.class);
 
-    enum Task { 
-        deployApp(DeployApp.class),
-        deploy(Deploy.class), 
-        node(Main.class), 
-        addNodes(AddNodes.class), 
-        zkServer(ZKServer.class), 
-        newCluster(CreateCluster.class), 
-        adapter(null), 
-        genericAdapter(GenericEventAdapter.class),
-        newApp(CreateApp.class), 
-        s4r(Package.class), 
-        status(S4Status.class),
-        createTask(CreateTask.class), 
-        rebalanceTask(RebalanceTask.class);
+    enum Task {
+        deployApp(DeployApp.class), deploy(Deploy.class), node(Main.class), addNodes(AddNodes.class), zkServer(
+                ZKServer.class), newCluster(CreateCluster.class), adapter(null), genericAdapter(
+                GenericEventAdapter.class), newApp(CreateApp.class), s4r(Package.class), status(S4Status.class), createTask(
+                CreateTask.class), rebalanceTask(RebalanceTask.class);
 
         Class<?> target;