ZOOKEEPER-82. Make the ZooKeeperServer more DI friendly. (Hiram Chirino via mahadev)

git-svn-id: https://svn.apache.org/repos/asf/hadoop/zookeeper/trunk@686284 13f79535-47bb-0310-9956-ffa450edef68

README.txt

@@ -11,7 +11,7 @@ Starting the server:
 
 2) start the server with the following comand line:
 
-java -cp zookeeper-dev.jar:java/lib/log4j-1.2.15.jar org.apache.zookeeper.server.quorum.QuorumPeer conf/zoo.cfg
+java -cp conf:zookeeper-dev.jar:src/java/lib/log4j-1.2.15.jar org.apache.zookeeper.server.quorum.QuorumPeerMain conf/zoo.cfg
 
 Notice that the server is picking up the log4j.properties file from the conf directory (default).
 
@@ -21,8 +21,8 @@ Starting a client shell
 
 1) run the following command
 
-java -cp zookeeper-dev.jar:java/lib/log4j-1.2.15.jar org.apache.zookeeper.ZooKeeper <server>:<port>
+java -cp conf:zookeeper-dev.jar:src/java/lib/log4j-1.2.15.jar org.apache.zookeeper.ZooKeeperMain <server>:<port>
 
 where server and port correspond to the ZooKeeper configuration.
 
-Notice that the client is picking up the log4j_console.properties file from the conf directory (default).
+Notice that the client is picking up the log4j.properties file from the conf directory (default).

bin/zkCli.sh

@@ -34,4 +34,4 @@ eval `grep -e "^dataDir=" $ZOOCFG`
 
 java "-Dzookeeper.log.dir=${ZOO_LOG_DIR}" "-Dzookeeper.root.logger=${ZOO_LOG4J_PROP}" \
      -cp $CLASSPATH $JVMFLAGS \
-     org.apache.zookeeper.ZooKeeper $@
+     org.apache.zookeeper.ZooKeeperMain $@

bin/zkServer.sh

@@ -30,7 +30,7 @@ case $1 in
 start) 
     echo -n "Starting zookeeper ... "
     java  "-Dzookeeper.log.dir=${ZOO_LOG_DIR}" "-Dzookeeper.root.logger=${ZOO_LOG4J_PROP}" \
-    -cp $CLASSPATH $JVMFLAGS org.apache.zookeeper.server.quorum.QuorumPeer $ZOOCFG &
+    -cp $CLASSPATH $JVMFLAGS org.apache.zookeeper.server.quorum.QuorumPeerMain $ZOOCFG &
     echo STARTED
     ;;
 stop) 

src/java/jmx/org/apache/zookeeper/jmx/server/ZooKeeperServerBean.java

@@ -24,21 +24,32 @@ import java.util.Date;
 
 import org.apache.zookeeper.Version;
 import org.apache.zookeeper.jmx.ZKMBeanInfo;
-import org.apache.zookeeper.server.ServerConfig;
 import org.apache.zookeeper.server.ServerStats;
+import org.apache.zookeeper.server.ZooKeeperServer;
 
 /**
  * This class implements the zookeeper server MBean interface.
  */
 public class ZooKeeperServerBean implements ZooKeeperServerMXBean, ZKMBeanInfo {
     private Date startTime=new Date();
+    private ZooKeeperServer zooKeeperServer;
+    
+    public ZooKeeperServerBean() {        
+    }
+    public ZooKeeperServerBean(ZooKeeperServer zooKeeperServer) {
+        this.zooKeeperServer = zooKeeperServer;
+    }
 
     public String getClientPort() {
+        ZooKeeperServer zks = getZooKeeperServer();
+        if( zks == null ) {
+            return null;
+        }
         try {
             return InetAddress.getLocalHost().getHostAddress() + ":"
-                    + ServerConfig.getClientPort();
+                    + zks.getClientPort();
         } catch (UnknownHostException e) {
-            return "localhost:" + ServerConfig.getClientPort();
+            return "localhost:" + zks.getClientPort();
         }
     }
     
@@ -94,4 +105,13 @@ public class ZooKeeperServerBean implements ZooKeeperServerMXBean, ZKMBeanInfo {
         ServerStats.getInstance().resetRequestCounters();
         ServerStats.getInstance().resetLatency();
     }
+
+    public ZooKeeperServer getZooKeeperServer() {
+        return zooKeeperServer;
+    }
+
+    public void setZooKeeperServer(ZooKeeperServer zooKeeperServer) {
+        this.zooKeeperServer = zooKeeperServer;
+    }
+
 }

src/java/jmx/org/apache/zookeeper/jmx/server/quorum/LocalPeerBean.java

@@ -19,6 +19,7 @@
 package org.apache.zookeeper.jmx.server.quorum;
 
 import org.apache.zookeeper.jmx.server.ZooKeeperServerBean;
+import org.apache.zookeeper.server.ZooKeeperServer;
 import org.apache.zookeeper.server.quorum.QuorumPeer;
 
 /**
@@ -45,4 +46,8 @@ public class LocalPeerBean extends ZooKeeperServerBean implements LocalPeerMXBea
             peer.getQuorumAddress().getPort();
     }
 
+    @Override
+    public ZooKeeperServer getZooKeeperServer() {
+        return peer.getActiveServer();
+    }
 }

src/java/jmx/org/apache/zookeeper/server/ManagedZooKeeperServer.java

@@ -28,6 +28,7 @@ import org.apache.zookeeper.jmx.MBeanRegistry;
 import org.apache.zookeeper.jmx.server.ConnectionBean;
 import org.apache.zookeeper.jmx.server.DataTreeBean;
 import org.apache.zookeeper.jmx.server.ZooKeeperServerBean;
+import org.apache.zookeeper.server.ZooKeeperServer.DataTreeBuilder;
 import org.apache.zookeeper.server.util.ConnectionObserver;
 import org.apache.zookeeper.server.util.ObserverManager;
 import org.apache.zookeeper.server.util.ServerObserver;
@@ -71,7 +72,7 @@ public class ManagedZooKeeperServer extends ObservableZooKeeperServer {
 
         public void onStartup(ZooKeeperServer server) {
             try {
-                svrBean = new ZooKeeperServerBean();
+                svrBean = new ZooKeeperServerBean(server);
                 MBeanRegistry.getInstance().register(svrBean, null);
                 dataTreeBean = new DataTreeBean(server.dataTree);
                 MBeanRegistry.getInstance().register(dataTreeBean, svrBean);
@@ -104,37 +105,22 @@ public class ManagedZooKeeperServer extends ObservableZooKeeperServer {
         }
     }
 
-    public ManagedZooKeeperServer(File dataDir, File dataLogDir, 
-            int tickTime,DataTreeBuilder treeBuilder) throws IOException {
-        super(dataDir, dataLogDir, tickTime,treeBuilder);
+    public ManagedZooKeeperServer() {
+        super();
         ObserverManager.getInstance().add(new ManagedServerObserver());
         ObserverManager.getInstance().add(new ManagedConnectionObserver());
     }
 
-    public ManagedZooKeeperServer(DataTreeBuilder treeBuilder) throws IOException {
-        super(treeBuilder);
+    public ManagedZooKeeperServer(File dataDir, File dataLogDir, int tickTime, DataTreeBuilder treeBuilder) throws IOException {
+        super(dataDir, dataLogDir, tickTime, treeBuilder);
         ObserverManager.getInstance().add(new ManagedServerObserver());
         ObserverManager.getInstance().add(new ManagedConnectionObserver());
     }
 
-    /**
-     * To start the server specify the client port number and the data directory
-     * on the command line.
-     * @see ServerConfig#parse(String[])
-     * @param args command line parameters.
-     */
-    public static void main(String[] args) {
-        ServerConfig.parse(args);
-        ZooKeeperObserverManager.setAsConcrete();
-        runStandalone(new Factory() {
-            public NIOServerCnxn.Factory createConnectionFactory()throws IOException {
-                return new ObservableNIOServerCnxn.Factory(getClientPort());
-            }
-            public ZooKeeperServer createServer() throws IOException {
-                // TODO: we may want to build an observable/managed data tree here instead
-                return new ManagedZooKeeperServer(new BasicDataTreeBuilder());
-            }
-        });
+    public ManagedZooKeeperServer(File dataDir, File dataLogDir, int tickTime) throws IOException {
+        super(dataDir, dataLogDir, tickTime);
+        ObserverManager.getInstance().add(new ManagedServerObserver());
+        ObserverManager.getInstance().add(new ManagedConnectionObserver());
     }
 
 }

src/java/jmx/org/apache/zookeeper/server/ManagedZooKeeperServerMain.java

@@ -0,0 +1,80 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.server;
+
+import static org.apache.zookeeper.server.ServerConfig.getClientPort;
+
+import java.io.File;
+import java.io.IOException;
+
+import org.apache.log4j.Logger;
+import org.apache.zookeeper.jmx.server.ConnectionMXBean;
+import org.apache.zookeeper.jmx.server.DataTreeMXBean;
+import org.apache.zookeeper.jmx.server.ZooKeeperServerMXBean;
+import org.apache.zookeeper.server.util.ZooKeeperObserverManager;
+
+/**
+ * This class launches a standalone zookeeper server with JMX support
+ * enabled. The users can connect to the server JVM and manage the server state 
+ * (such as currently open client connections) and view runtime statistics using 
+ * one of existing GUI JMX consoles (jconsole, for example). Please refer to 
+ * the JDK vendor documentation for further information on how to enable JMX 
+ * support in the JVM.
+ * <p>
+ * The server provides following MBeans:
+ * <ul>
+ *  <li>Zookeeper server MBean -- provides various configuraton data and runtime 
+ *  statistics, see {@link ZooKeeperServerMXBean}
+ *  <li>Data tree MBean -- provides runtime data tree statistics, see 
+ *  {@link DataTreeMXBean}
+ *  <li>Client connection MBean -- provides runtime statistics as well as 
+ *  connection management operations, see {@link ConnectionMXBean}
+ * </ul>
+ * The client connection is a dynamic resource and therefore the connection
+ * MBeans are dynamically created and destroyed as the clients connect to and 
+ * disconnect from the server.
+ */
+public class ManagedZooKeeperServerMain extends ZooKeeperServerMain {
+    
+    /**
+     * To start the server specify the client port number and the data directory
+     * on the command line.
+     * @see ServerConfig#parse(String[])
+     * @param args command line parameters.
+     */
+    public static void main(String[] args) {
+        ServerConfig.parse(args);
+        ZooKeeperObserverManager.setAsConcrete();
+        runStandalone(new ZooKeeperServer.Factory() {
+            public NIOServerCnxn.Factory createConnectionFactory()throws IOException {
+                return new ObservableNIOServerCnxn.Factory(getClientPort());
+            }
+            public ZooKeeperServer createServer() throws IOException {
+                ManagedZooKeeperServer zks = new ManagedZooKeeperServer();
+                zks.setDataDir(new File(ServerConfig.getDataDir()));
+                zks.setDataLogDir(new File(ServerConfig.getDataLogDir()));
+                zks.setClientPort(ServerConfig.getClientPort());
+                // TODO: we may want to build an observable/managed data tree here instead
+                zks.setTreeBuilder(new ZooKeeperServer.BasicDataTreeBuilder());
+                return zks;
+            }
+        });
+    }
+
+}

src/java/jmx/org/apache/zookeeper/server/ObservableZooKeeperServer.java

@@ -20,6 +20,8 @@ package org.apache.zookeeper.server;
 
 import java.io.File;
 import java.io.IOException;
+
+import org.apache.zookeeper.server.ZooKeeperServer.DataTreeBuilder;
 /**
  * The observable server broadcast notifications when its state changes. 
  * 
@@ -31,13 +33,16 @@ public class ObservableZooKeeperServer extends ZooKeeperServer{
 
     private ZooKeeperObserverNotifier notifier=new ZooKeeperObserverNotifier(this);
     
-    public ObservableZooKeeperServer(File dataDir, File dataLogDir, 
-            int tickTime,DataTreeBuilder treeBuilder) throws IOException {
-        super(dataDir, dataLogDir, tickTime,treeBuilder);
+    public ObservableZooKeeperServer() {
+        super();
+    }
+
+    public ObservableZooKeeperServer(File dataDir, File dataLogDir, int tickTime, DataTreeBuilder treeBuilder) throws IOException {
+        super(dataDir, dataLogDir, tickTime, treeBuilder);
     }
 
-    public ObservableZooKeeperServer(DataTreeBuilder treeBuilder) throws IOException {
-        super(treeBuilder);
+    public ObservableZooKeeperServer(File dataDir, File dataLogDir, int tickTime) throws IOException {
+        super(dataDir, dataLogDir, tickTime);
     }
 
     public void shutdown() {

src/java/jmx/org/apache/zookeeper/server/quorum/ManagedQuorumPeer.java

@@ -44,6 +44,7 @@ import org.apache.zookeeper.server.ObservableNIOServerCnxn;
 import org.apache.zookeeper.server.ServerCnxn;
 import org.apache.zookeeper.server.ZooKeeperServer;
 import org.apache.zookeeper.server.ZooTrace;
+import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
 import org.apache.zookeeper.server.util.ConnectionObserver;
 import org.apache.zookeeper.server.util.ObserverManager;
 import org.apache.zookeeper.server.util.QuorumPeerObserver;
@@ -201,45 +202,21 @@ public class ManagedQuorumPeer extends ObservableQuorumPeer {
         ObserverManager.getInstance().add(new ManagedConnectionObserver());        
     }
     
-    public ManagedQuorumPeer(ArrayList<QuorumServer> quorumPeers, File dataDir,
-            File dataLogDir,int electionAlg, int electionPort,long myid,    int tickTime, 
-            int initLimit, int syncLimit,NIOServerCnxn.Factory cnxnFactory) 
-                throws IOException {
-        super(quorumPeers, dataDir, dataLogDir,electionAlg, electionPort,myid,
-                tickTime, initLimit, syncLimit,cnxnFactory);
+    public ManagedQuorumPeer() {
+        super();
         setupObservers();
     }
 
-    public ManagedQuorumPeer(NIOServerCnxn.Factory cnxnFactory) throws IOException {
-        super(cnxnFactory);
+    public ManagedQuorumPeer(ArrayList<QuorumServer> quorumPeers, File dataDir, File dataLogDir, int clientPort, int electionAlg, int electionPort, long myid, int tickTime, int initLimit,
+                                int syncLimit) throws IOException {
+        super(quorumPeers, dataDir, dataLogDir, clientPort, electionAlg, electionPort, myid, tickTime, initLimit, syncLimit);
         setupObservers();
     }
 
-    /**
-     * To start the replicated server specify the configuration file name on the
-     * command line.
-     * @param args command line
-     */
-    public static void main(String[] args) {
-        if (args.length == 2) {
-            ManagedZooKeeperServer.main(args);
-            return;
-        }
-        QuorumPeerConfig.parse(args);
-        if (!QuorumPeerConfig.isStandalone()) {
-            ZooKeeperObserverManager.setAsConcrete();
-            runPeer(new QuorumPeer.Factory() {
-                public QuorumPeer create(NIOServerCnxn.Factory cnxnFactory)
-                        throws IOException {
-                    return new ManagedQuorumPeer(cnxnFactory);
-                }
-                public NIOServerCnxn.Factory createConnectionFactory() throws IOException {
-                    return new ObservableNIOServerCnxn.Factory(getClientPort());
-                }
-            });
-        }else{
-            // there is only server in the quorum -- run as standalone
-            ManagedZooKeeperServer.main(args);
-        }
+    public ManagedQuorumPeer(ArrayList<QuorumServer> quorumPeers, File dataDir, File dataLogDir, int electionType, int electionPort, long myid, int tickTime, int initLimit, int syncLimit,
+                                NIOServerCnxn.Factory cnxnFactory) throws IOException {
+        super(quorumPeers, dataDir, dataLogDir, electionType, electionPort, myid, tickTime, initLimit, syncLimit, cnxnFactory);
+        setupObservers();
     }
+
 }

src/java/jmx/org/apache/zookeeper/server/quorum/ManagedQuorumPeerMain.java

@@ -0,0 +1,142 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.server.quorum;
+
+import static org.apache.zookeeper.server.ServerConfig.getClientPort;
+import static org.apache.zookeeper.server.quorum.QuorumPeerConfig.getElectionAlg;
+import static org.apache.zookeeper.server.quorum.QuorumPeerConfig.getServerId;
+import static org.apache.zookeeper.server.quorum.QuorumPeerConfig.getServers;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.concurrent.ConcurrentHashMap;
+
+import org.apache.log4j.Logger;
+
+import org.apache.zookeeper.jmx.MBeanRegistry;
+import org.apache.zookeeper.jmx.ZKMBeanInfo;
+import org.apache.zookeeper.jmx.server.ConnectionBean;
+import org.apache.zookeeper.jmx.server.DataTreeBean;
+import org.apache.zookeeper.jmx.server.quorum.FollowerBean;
+import org.apache.zookeeper.jmx.server.quorum.LeaderBean;
+import org.apache.zookeeper.jmx.server.quorum.LeaderElectionBean;
+import org.apache.zookeeper.jmx.server.quorum.LocalPeerBean;
+import org.apache.zookeeper.jmx.server.quorum.QuorumBean;
+import org.apache.zookeeper.jmx.server.quorum.RemotePeerBean;
+import org.apache.zookeeper.jmx.server.quorum.ServerBean;
+import org.apache.zookeeper.server.ManagedZooKeeperServerMain;
+import org.apache.zookeeper.server.ManagedZooKeeperServer;
+import org.apache.zookeeper.server.NIOServerCnxn;
+import org.apache.zookeeper.server.ObservableNIOServerCnxn;
+import org.apache.zookeeper.server.ServerCnxn;
+import org.apache.zookeeper.server.ServerConfig;
+import org.apache.zookeeper.server.ZooKeeperServer;
+import org.apache.zookeeper.server.ZooTrace;
+import org.apache.zookeeper.server.util.ConnectionObserver;
+import org.apache.zookeeper.server.util.ObserverManager;
+import org.apache.zookeeper.server.util.QuorumPeerObserver;
+import org.apache.zookeeper.server.util.ServerObserver;
+import org.apache.zookeeper.server.util.ZooKeeperObserverManager;
+
+/**
+ * This class launches a replicated zookeeper server with JMX support
+ * enabled. The users can connect to the server JVM and manage 
+ * the server state (such as currently open client connections) and view runtime
+ * statistics using one of existing GUI JMX consoles (jconsole, for example).
+ * Please refer to the JDK vendor documentation for further information on how
+ * to enable JMX support in the JVM.
+ * <p>
+ * The server provides following MBeans:
+ * <ul>
+ *  <li>Quorum MBean -- provides quorum runtime statistics, see {@link QuorumMXBean}.
+ *  <li>Peer MBean -- provides information about quorum peers (local and remote),
+ *  see {@link LocalPeerMXBean} and {@link RemotePeerMXBean}.
+ *  <li>Leader election MBean -- provides runtime info on leader election protocol,
+ *  see {@link LeaderElectionMXBean}
+ *  <li>Zookeeper server MBean -- provides various configuraton data and runtime 
+ *  statistics, see {@link ZooKeeperServerMXBean}
+ *  <li>Data tree MBean -- provides runtime data tree statistics, see 
+ *  {@link DataTreeMXBean}
+ *  <li>Client connection MBean -- provides runtime statistics as well as 
+ *  connection management operations, see {@link ConnectionMXBean}
+ * </ul>
+ * The client connection is a dynamic resource and therefore the connection
+ * MBeans are dynamically created and destroyed as the clients connect to and 
+ * disconnect from the server.
+ */
+public class ManagedQuorumPeerMain {
+    
+    private static final Logger LOG = Logger.getLogger(ManagedQuorumPeerMain.class);
+
+    /**
+     * To start the replicated server specify the configuration file name on the
+     * command line.
+     * @param args command line
+     */
+    public static void main(String[] args) {
+        if (args.length == 2) {
+            ManagedZooKeeperServerMain.main(args);
+            return;
+        }
+        QuorumPeerConfig.parse(args);
+        if (!QuorumPeerConfig.isStandalone()) {
+            ZooKeeperObserverManager.setAsConcrete();
+            runPeer(new QuorumPeer.Factory() {
+                public QuorumPeer create(NIOServerCnxn.Factory cnxnFactory)
+                        throws IOException {
+                    
+                    ManagedQuorumPeer peer = new ManagedQuorumPeer();
+                    peer.setClientPort(ServerConfig.getClientPort());
+                    peer.setDataDir(new File(ServerConfig.getDataDir()));
+                    peer.setDataLogDir(new File(ServerConfig.getDataLogDir()));
+                    peer.setQuorumPeers(QuorumPeerConfig.getServers());
+                    peer.setElectionPort(QuorumPeerConfig.getElectionPort());
+                    peer.setElectionType(QuorumPeerConfig.getElectionAlg());
+                    peer.setMyid(QuorumPeerConfig.getServerId());
+                    peer.setTickTime(QuorumPeerConfig.getTickTime());
+                    peer.setInitLimit(QuorumPeerConfig.getInitLimit());
+                    peer.setSyncLimit(QuorumPeerConfig.getSyncLimit());
+                    peer.setCnxnFactory(cnxnFactory);
+                    return peer;
+                    
+                }
+                public NIOServerCnxn.Factory createConnectionFactory() throws IOException {
+                    return new ObservableNIOServerCnxn.Factory(getClientPort());
+                }
+            });
+        }else{
+            // there is only server in the quorum -- run as standalone
+            ManagedZooKeeperServerMain.main(args);
+        }
+    }
+    
+    public static void runPeer(QuorumPeer.Factory qpFactory) {
+        try {
+            QuorumStats.registerAsConcrete();
+            QuorumPeer self = qpFactory.create(qpFactory.createConnectionFactory());
+            self.start();
+            self.join();
+        } catch (Exception e) {
+            LOG.fatal("Unexpected exception",e);
+        }
+        System.exit(2);
+    }
+
+}

src/java/jmx/org/apache/zookeeper/server/quorum/ObservableQuorumPeer.java

@@ -24,6 +24,8 @@ import java.util.ArrayList;
 
 import org.apache.zookeeper.server.NIOServerCnxn;
 import org.apache.zookeeper.server.ZooKeeperServer;
+import org.apache.zookeeper.server.NIOServerCnxn.Factory;
+import org.apache.zookeeper.server.quorum.QuorumPeer.QuorumServer;
 import org.apache.zookeeper.server.util.EventInfo;
 import org.apache.zookeeper.server.util.ObservableComponent;
 import org.apache.zookeeper.server.util.ObserverManager;
@@ -58,17 +60,20 @@ public class ObservableQuorumPeer extends QuorumPeer implements ObservableCompon
         };
         public abstract void dispatch(ObservableQuorumPeer peer,QuorumPeerObserver ob);
     }
-    
-    public ObservableQuorumPeer(ArrayList<QuorumServer> quorumPeers,
-            File dataDir, File dataLogDir, int electionAlg,    int electionPort,long myid, 
-            int tickTime, int initLimit, int syncLimit,NIOServerCnxn.Factory cnxnFactory)
-            throws IOException {
-        super(quorumPeers, dataDir, dataLogDir,electionAlg,electionPort, myid,
-                tickTime, initLimit, syncLimit,cnxnFactory);
+
+
+    public ObservableQuorumPeer() {
+        super();
+    }
+
+    public ObservableQuorumPeer(ArrayList<QuorumServer> quorumPeers, File dataDir, File dataLogDir, int clientPort, int electionAlg, int electionPort, long myid, int tickTime, int initLimit,
+                                int syncLimit) throws IOException {
+        super(quorumPeers, dataDir, dataLogDir, clientPort, electionAlg, electionPort, myid, tickTime, initLimit, syncLimit);
     }
 
-    public ObservableQuorumPeer(NIOServerCnxn.Factory cnxnFactory) throws IOException {
-        super(cnxnFactory);
+    public ObservableQuorumPeer(ArrayList<QuorumServer> quorumPeers, File dataDir, File dataLogDir, int electionType, int electionPort, long myid, int tickTime, int initLimit, int syncLimit,
+                                NIOServerCnxn.Factory cnxnFactory) throws IOException {
+        super(quorumPeers, dataDir, dataLogDir, electionType, electionPort, myid, tickTime, initLimit, syncLimit, cnxnFactory);
     }
 
     // instantiate an observable follower

src/java/main/org/apache/zookeeper/ZooKeeper.java

@@ -18,11 +18,7 @@
 
 package org.apache.zookeeper;
 
-import java.io.BufferedReader;
 import java.io.IOException;
-import java.io.InputStreamReader;
-import java.util.ArrayList;
-import java.util.Date;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -30,16 +26,13 @@ import java.util.Map;
 import java.util.Set;
 
 import org.apache.log4j.Logger;
-
 import org.apache.zookeeper.AsyncCallback.ACLCallback;
 import org.apache.zookeeper.AsyncCallback.ChildrenCallback;
 import org.apache.zookeeper.AsyncCallback.DataCallback;
 import org.apache.zookeeper.AsyncCallback.StatCallback;
 import org.apache.zookeeper.AsyncCallback.StringCallback;
 import org.apache.zookeeper.AsyncCallback.VoidCallback;
-import org.apache.zookeeper.ZooDefs.Ids;
 import org.apache.zookeeper.data.ACL;
-import org.apache.zookeeper.data.Id;
 import org.apache.zookeeper.data.Stat;
 import org.apache.zookeeper.proto.CreateRequest;
 import org.apache.zookeeper.proto.CreateResponse;
@@ -195,7 +188,7 @@ public class ZooKeeper {
             }
         }
     }
-    
+
     /**
      * Register a watcher for a particular path.
      */
@@ -207,10 +200,10 @@ public class ZooKeeper {
                 Watcher watcher, String path)
         {
             this.watches = watches;
-            this.watcher = watcher;  
+            this.watcher = watcher;
             this.path = path;
         }
-        
+
         /**
          * Register the watcher with the set of watches on path.
          * @param rc the result code of the operation that attempted to
@@ -936,200 +929,6 @@ public class ZooKeeper {
 
     // Everything below this line is for testing!
 
-    static void usage() {
-        System.err.println("ZooKeeper host:port cmd args");
-        System.err.println("\tcreate path data acl");
-        System.err.println("\tdelete path [version]");
-        System.err.println("\tset path data [version]");
-        System.err.println("\tget path [watch]");
-        System.err.println("\tls path [watch]");
-        System.err.println("\tgetAcl path");
-        System.err.println("\tsetAcl path acl");
-        System.err.println("\tstat path [watch]");
-        System.err.println("\tsync path");
-    }
-
-    static private class MyWatcher implements Watcher {
-        public void process(WatcherEvent event) {
-            System.err.println(event.getPath() + ": " + event.getState() + "-"
-                    + event.getType());
-        }
-    }
-
-    static private int getPermFromString(String permString) {
-        int perm = 0;
-        for (int i = 0; i < permString.length(); i++) {
-            switch (permString.charAt(i)) {
-            case 'r':
-                perm |= ZooDefs.Perms.READ;
-                break;
-            case 'w':
-                perm |= ZooDefs.Perms.WRITE;
-                break;
-            case 'c':
-                perm |= ZooDefs.Perms.CREATE;
-                break;
-            case 'd':
-                perm |= ZooDefs.Perms.DELETE;
-                break;
-            case 'a':
-                perm |= ZooDefs.Perms.ADMIN;
-                break;
-            default:
-                System.err
-                        .println("Unknown perm type: " + permString.charAt(i));
-            }
-        }
-        return perm;
-    }
-
-    private static void printStat(Stat stat) {
-        System.err.println("ctime = " + new Date(stat.getCtime()).toString());
-        System.err.println("ctime = " + new Date(stat.getMtime()).toString());
-        System.err.println("cversion = " + stat.getCversion());
-        System.err.println("cZxid = " + stat.getCzxid());
-        System.err.println("mZxid = " + stat.getMzxid());
-        System.err.println("dataVersion = " + stat.getVersion());
-        System.err.println("aclVersion = " + stat.getAversion());
-    }
-
-    public static void main(String args[]) throws NumberFormatException,
-            KeeperException, IOException, InterruptedException {
-        if (args.length == 1) {
-            ZooKeeper zooKeeper = new ZooKeeper(args[0], 5000, new MyWatcher());
-            BufferedReader br = new BufferedReader(new InputStreamReader(
-                    System.in));
-            String line;
-            while ((line = br.readLine()) != null) {
-                line = "ignore " + line;
-                args = line.split(" ");
-                processCmd(args, zooKeeper);
-            }
-        } else if (args.length < 3) {
-            usage();
-        }
-
-        ZooKeeper zooKeeper = new ZooKeeper(args[0], 5000, new MyWatcher());
-        boolean watch = processCmd(args, zooKeeper);
-        if (!watch) {
-            System.exit(0);
-        }
-    }
-
-    private static DataCallback dataCallback = new DataCallback() {
-
-        public void processResult(int rc, String path, Object ctx, byte[] data,
-                Stat stat) {
-            System.out.println("rc = " + rc + " path = " + path + " data = "
-                    + (data == null ? "null" : new String(data)) + " stat = ");
-            printStat(stat);
-        }
-
-    };
-
-    private static boolean processCmd(String[] args, ZooKeeper zooKeeper)
-            throws KeeperException, IOException, InterruptedException {
-        Stat stat = new Stat();
-        if (args.length < 2) {
-            return false;
-        }
-        if (args.length < 3) {
-            usage();
-            return false;
-        }
-        String cmd = args[1];
-        boolean watch = args.length > 3;
-        String path = args[2];
-        List<ACL> acl = Ids.OPEN_ACL_UNSAFE;
-        System.out.println("Processing " + cmd);
-        try {
-            if (cmd.equals("create") && args.length >= 4) {
-                if (args.length == 5) {
-                    acl = parseACLs(args[4]);
-                }
-                String newPath = zooKeeper.create(path, args[3].getBytes(), acl, 0);
-                System.err.println("Created " + newPath);
-            } else if (cmd.equals("delete") && args.length >= 3) {
-                zooKeeper.delete(path, watch ? Integer.parseInt(args[3]) : -1);
-            } else if (cmd.equals("set") && args.length >= 4) {
-                stat = zooKeeper.setData(path, args[3].getBytes(),
-                        args.length > 4 ? Integer.parseInt(args[4]) : -1);
-                printStat(stat);
-            } else if (cmd.equals("aget") && args.length >= 3) {
-                zooKeeper.getData(path, watch, dataCallback, path);
-            } else if (cmd.equals("get") && args.length >= 3) {
-                byte data[] = zooKeeper.getData(path, watch, stat);
-                System.out.println(new String(data));
-                printStat(stat);
-            } else if (cmd.equals("ls") && args.length >= 3) {
-                List<String> children = zooKeeper.getChildren(path, watch);
-                System.out.println(children);
-            } else if (cmd.equals("getAcl") && args.length >= 2) {
-                acl = zooKeeper.getACL(path, stat);
-                for (ACL a : acl) {
-                    System.out.println(a.getId() + ": "
-                            + getPermString(a.getPerms()));
-                }
-            } else if (cmd.equals("setAcl") && args.length >= 4) {
-
-                stat = zooKeeper.setACL(path, parseACLs(args[3]),
-                        args.length > 4 ? Integer.parseInt(args[4]) : -1);
-                printStat(stat);
-            } else if (cmd.equals("stat") && args.length >= 3) {
-                stat = zooKeeper.exists(path, watch);
-                printStat(stat);
-            } else {
-                usage();
-            }
-
-            return watch;
-        } catch (KeeperException e) {
-            System.err.println(e.getClass().getName() + ": " + e.getMessage());
-            return false;
-	}
-    }
-
-    private static String getPermString(int perms) {
-        StringBuffer p = new StringBuffer();
-        if ((perms & ZooDefs.Perms.CREATE) != 0) {
-            p.append('c');
-        }
-        if ((perms & ZooDefs.Perms.DELETE) != 0) {
-            p.append('d');
-        }
-        if ((perms & ZooDefs.Perms.READ) != 0) {
-            p.append('r');
-        }
-        if ((perms & ZooDefs.Perms.WRITE) != 0) {
-            p.append('w');
-        }
-        if ((perms & ZooDefs.Perms.ADMIN) != 0) {
-            p.append('a');
-        }
-        return p.toString();
-    }
-
-    private static List<ACL> parseACLs(String aclString) {
-        List<ACL> acl;
-        String acls[] = aclString.split(",");
-        acl = new ArrayList<ACL>();
-        for (String a : acls) {
-            int firstColon = a.indexOf(':');
-            int lastColon = a.lastIndexOf(':');
-            if (firstColon == -1 || lastColon == -1 || firstColon == lastColon) {
-                System.err
-                        .println(a + " does not have the form scheme:id:perm");
-                continue;
-            }
-            ACL newAcl = new ACL();
-            newAcl.setId(new Id(a.substring(0, firstColon), a.substring(
-                    firstColon + 1, lastColon)));
-            newAcl.setPerms(getPermFromString(a.substring(lastColon + 1)));
-            acl.add(newAcl);
-        }
-        return acl;
-    }
-
     public void disconnect() throws IOException {
         cnxn.close();
     }

src/java/main/org/apache/zookeeper/ZooKeeperMain.java

@@ -0,0 +1,228 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.zookeeper;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+
+import org.apache.zookeeper.AsyncCallback.DataCallback;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.apache.zookeeper.data.ACL;
+import org.apache.zookeeper.data.Id;
+import org.apache.zookeeper.data.Stat;
+import org.apache.zookeeper.proto.WatcherEvent;
+
+/**
+ * The command line client to ZooKeeper.
+ * 
+ */
+public class ZooKeeperMain {
+
+    static void usage() {
+        System.err.println("ZooKeeper host:port cmd args");
+        System.err.println("\tcreate path data acl");
+        System.err.println("\tdelete path [version]");
+        System.err.println("\tset path data [version]");
+        System.err.println("\tget path [watch]");
+        System.err.println("\tls path [watch]");
+        System.err.println("\tgetAcl path");
+        System.err.println("\tsetAcl path acl");
+        System.err.println("\tstat path [watch]");
+        System.err.println("\tsync path");
+    }
+
+    static private class MyWatcher implements Watcher {
+        public void process(WatcherEvent event) {
+            System.err.println(event.getPath() + ": " + event.getState() + "-"
+                    + event.getType());
+        }
+    }
+
+    static private int getPermFromString(String permString) {
+        int perm = 0;
+        for (int i = 0; i < permString.length(); i++) {
+            switch (permString.charAt(i)) {
+            case 'r':
+                perm |= ZooDefs.Perms.READ;
+                break;
+            case 'w':
+                perm |= ZooDefs.Perms.WRITE;
+                break;
+            case 'c':
+                perm |= ZooDefs.Perms.CREATE;
+                break;
+            case 'd':
+                perm |= ZooDefs.Perms.DELETE;
+                break;
+            case 'a':
+                perm |= ZooDefs.Perms.ADMIN;
+                break;
+            default:
+                System.err
+                        .println("Unknown perm type: " + permString.charAt(i));
+            }
+        }
+        return perm;
+    }
+
+    private static void printStat(Stat stat) {
+        System.err.println("ctime = " + new Date(stat.getCtime()).toString());
+        System.err.println("ctime = " + new Date(stat.getMtime()).toString());
+        System.err.println("cversion = " + stat.getCversion());
+        System.err.println("cZxid = " + stat.getCzxid());
+        System.err.println("mZxid = " + stat.getMzxid());
+        System.err.println("dataVersion = " + stat.getVersion());
+        System.err.println("aclVersion = " + stat.getAversion());
+    }
+
+    public static void main(String args[]) throws NumberFormatException,
+            KeeperException, IOException, InterruptedException {
+        if (args.length == 1) {
+            ZooKeeper zooKeeper = new ZooKeeper(args[0], 5000, new MyWatcher());
+            BufferedReader br = new BufferedReader(new InputStreamReader(
+                    System.in));
+            String line;
+            while ((line = br.readLine()) != null) {
+                line = "ignore " + line;
+                args = line.split(" ");
+                processCmd(args, zooKeeper);
+            }
+        } else if (args.length < 3) {
+            usage();
+        }
+
+        ZooKeeper zooKeeper = new ZooKeeper(args[0], 5000, new MyWatcher());
+        boolean watch = processCmd(args, zooKeeper);
+        if (!watch) {
+            System.exit(0);
+        }
+    }
+
+    private static DataCallback dataCallback = new DataCallback() {
+
+        public void processResult(int rc, String path, Object ctx, byte[] data,
+                Stat stat) {
+            System.out.println("rc = " + rc + " path = " + path + " data = "
+                    + (data == null ? "null" : new String(data)) + " stat = ");
+            printStat(stat);
+        }
+
+    };
+
+    private static boolean processCmd(String[] args, ZooKeeper zooKeeper)
+            throws KeeperException, IOException, InterruptedException {
+        Stat stat = new Stat();
+        if (args.length < 2) {
+            return false;
+        }
+        if (args.length < 3) {
+            usage();
+            return false;
+        }
+        String cmd = args[1];
+        boolean watch = args.length > 3;
+        String path = args[2];
+        List<ACL> acl = Ids.OPEN_ACL_UNSAFE;
+        System.out.println("Processing " + cmd);
+        if (cmd.equals("create") && args.length >= 4) {
+            if (args.length == 5) {
+                acl = parseACLs(args[4]);
+            }
+            String newPath = zooKeeper.create(path, args[3].getBytes(), acl, 0);
+            System.err.println("Created " + newPath);
+        } else if (cmd.equals("delete") && args.length >= 3) {
+            zooKeeper.delete(path, watch ? Integer.parseInt(args[3]) : -1);
+        } else if (cmd.equals("set") && args.length >= 4) {
+            stat = zooKeeper.setData(path, args[3].getBytes(),
+                    args.length > 4 ? Integer.parseInt(args[4]) : -1);
+            printStat(stat);
+        } else if (cmd.equals("aget") && args.length >= 3) {
+            zooKeeper.getData(path, watch, dataCallback, path);
+        } else if (cmd.equals("get") && args.length >= 3) {
+            byte data[] = zooKeeper.getData(path, watch, stat);
+            System.out.println(new String(data));
+            printStat(stat);
+        } else if (cmd.equals("ls") && args.length >= 3) {
+            List<String> children = zooKeeper.getChildren(path, watch);
+            System.out.println(children);
+        } else if (cmd.equals("getAcl") && args.length >= 2) {
+            acl = zooKeeper.getACL(path, stat);
+            for (ACL a : acl) {
+                System.out.println(a.getId() + ": "
+                        + getPermString(a.getPerms()));
+            }
+        } else if (cmd.equals("setAcl") && args.length >= 4) {
+
+            stat = zooKeeper.setACL(path, parseACLs(args[3]),
+                    args.length > 4 ? Integer.parseInt(args[4]) : -1);
+            printStat(stat);
+        } else if (cmd.equals("stat") && args.length >= 3) {
+            stat = zooKeeper.exists(path, watch);
+            printStat(stat);
+        } else {
+            usage();
+        }
+        return watch;
+    }
+
+    private static String getPermString(int perms) {
+        StringBuffer p = new StringBuffer();
+        if ((perms & ZooDefs.Perms.CREATE) != 0) {
+            p.append('c');
+        }
+        if ((perms & ZooDefs.Perms.DELETE) != 0) {
+            p.append('d');
+        }
+        if ((perms & ZooDefs.Perms.READ) != 0) {
+            p.append('r');
+        }
+        if ((perms & ZooDefs.Perms.WRITE) != 0) {
+            p.append('w');
+        }
+        if ((perms & ZooDefs.Perms.ADMIN) != 0) {
+            p.append('a');
+        }
+        return p.toString();
+    }
+
+    private static List<ACL> parseACLs(String aclString) {
+        List<ACL> acl;
+        String acls[] = aclString.split(",");
+        acl = new ArrayList<ACL>();
+        for (String a : acls) {
+            int firstColon = a.indexOf(':');
+            int lastColon = a.lastIndexOf(':');
+            if (firstColon == -1 || lastColon == -1 || firstColon == lastColon) {
+                System.err
+                        .println(a + " does not have the form scheme:id:perm");
+                continue;
+            }
+            ACL newAcl = new ACL();
+            newAcl.setId(new Id(a.substring(0, firstColon), a.substring(
+                    firstColon + 1, lastColon)));
+            newAcl.setPerms(getPermFromString(a.substring(lastColon + 1)));
+            acl.add(newAcl);
+        }
+        return acl;
+    }
+
+}

src/java/main/org/apache/zookeeper/server/NIOServerCnxn.java

@@ -112,6 +112,10 @@ public class NIOServerCnxn implements Watcher, ServerCnxn {
             return (InetSocketAddress)ss.socket().getLocalSocketAddress();
         }
 
+        public int getLocalPort(){
+            return ss.socket().getLocalPort();
+        }
+
         private void addCnxn(NIOServerCnxn cnxn) {
             synchronized (cnxns) {
                 cnxns.add(cnxn);

src/java/main/org/apache/zookeeper/server/ZooKeeperServer.java

@@ -78,7 +78,7 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
     /**
      * Create an instance of Zookeeper server
      */
-    public interface Factory {
+    static public interface Factory {
         public ZooKeeperServer createServer() throws IOException;
 
         public NIOServerCnxn.Factory createConnectionFactory()
@@ -105,7 +105,7 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
     public int commitLogBuffer = 700;
     public LinkedList<Proposal> committedLog = new LinkedList<Proposal>();
     public long minCommittedLog, maxCommittedLog;
-    private DataTreeBuilder treeBuilder;
+    private DataTreeBuilder treeBuilder = new BasicDataTreeBuilder();
     public DataTree dataTree;
     protected SessionTracker sessionTracker;
     /**
@@ -130,49 +130,20 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
     int requestsInProcess;
     List<ChangeRecord> outstandingChanges = new ArrayList<ChangeRecord>();
     private NIOServerCnxn.Factory serverCnxnFactory;
+    private int clientPort;
 
-    /*
-     * Start up the ZooKeeper server.
-     *
-     * @param args the port and data directory
-     */
-    public static void main(String[] args) {
-        ServerConfig.parse(args);
-        runStandalone(new Factory() {
-            public NIOServerCnxn.Factory createConnectionFactory()
-                    throws IOException {
-                return new NIOServerCnxn.Factory(ServerConfig.getClientPort());
-            }
-
-            public ZooKeeperServer createServer() throws IOException {
-                return new ZooKeeperServer(new BasicDataTreeBuilder());
-            }
-        });
-    }
-
-    public static void runStandalone(Factory factory) {
-        try {
-            // Note that this thread isn't going to be doing anything else,
-            // so rather than spawning another thread, we will just call
-            // run() in this thread.
-            ServerStats.registerAsConcrete();
-            ZooKeeperServer zk = factory.createServer();
-            zk.startup();
-            NIOServerCnxn.Factory t = factory.createConnectionFactory();
-            t.setZooKeeperServer(zk);
-            t.join();
-            if (zk.isRunning())
-                zk.shutdown();
-        } catch (Exception e) {
-            LOG.fatal("Unexpected exception",e);
-        }
-        System.exit(0);
-    }
-
+ 
     void removeCnxn(ServerCnxn cnxn) {
         dataTree.removeCnxn(cnxn);
     }
 
+    /**
+     * 
+     * @throws IOException
+     */
+    public ZooKeeperServer() {
+        ServerStats.getInstance().setStatsProvider(this);
+    }
     /**
      * Creates a ZooKeeperServer instance. It sets everything up, but doesn't
      * actually start listening for clients until run() is invoked.
@@ -187,9 +158,6 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
         this.dataDir = dataDir;
         this.dataLogDir = dataLogDir;
         this.tickTime = tickTime;
-        if (!dataDir.isDirectory()) {
-            throw new IOException("data directory does not exist");
-        }
         ServerStats.getInstance().setStatsProvider(this);
     }
 
@@ -199,24 +167,10 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
      */
     public ZooKeeperServer(File dataDir, File dataLogDir, int tickTime)
             throws IOException {
-        this.treeBuilder = new BasicDataTreeBuilder();
+        this();
         this.dataDir = dataDir;
         this.dataLogDir = dataLogDir;
         this.tickTime = tickTime;
-        if (!dataDir.isDirectory()) {
-            throw new IOException("data directory does not exist");
-        }
-        ServerStats.getInstance().setStatsProvider(this);
-    }
-
-    /**
-     * Default constructor, relies on the config for its agrument values
-     *
-     * @throws IOException
-     */
-    public ZooKeeperServer(DataTreeBuilder treeBuilder) throws IOException {
-        this(new File(ServerConfig.getDataDir()), new File(ServerConfig
-                .getDataLogDir()), DEFAULT_TICK_TIME, treeBuilder);
     }
 
     public static long getZxidFromName(String name, String prefix) {
@@ -933,4 +887,59 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
     public long getOutstandingRequests() {
         return getInProcess();
     }
+
+    public int getTickTime() {
+        return tickTime;
+    }
+
+    public void setTickTime(int tickTime) {
+        this.tickTime = tickTime;
+    }
+
+    public DataTreeBuilder getTreeBuilder() {
+        return treeBuilder;
+    }
+
+    public void setTreeBuilder(DataTreeBuilder treeBuilder) {
+        this.treeBuilder = treeBuilder;
+    }
+
+    /**
+     * Gets directory for storing the snapshot
+     */
+    public File getDataDir() {
+        return dataDir;
+    }
+
+    /**
+     * Sets directory for storing the snapshot
+     */
+    public void setDataDir(File dataDir) throws IOException {
+        this.dataDir = dataDir;
+        if (!dataDir.isDirectory()) {
+            throw new IOException("data directory does not exist");
+        }
+    }
+
+    /**
+     * Gets directoy for storing the log tnxns
+     */
+    public File getDataLogDir() {
+        return dataLogDir;
+    }
+
+    /**
+     * Sets directoy for storing the log tnxns
+     */
+    public void setDataLogDir(File dataLogDir) {
+        this.dataLogDir = dataLogDir;
+    }
+
+    public int getClientPort() {
+        return clientPort;
+    }
+
+    public void setClientPort(int clientPort) {
+        this.clientPort = clientPort;
+    }
 }

src/java/main/org/apache/zookeeper/server/ZooKeeperServerMain.java

@@ -0,0 +1,74 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.zookeeper.server;
+
+import java.io.File;
+import java.io.IOException;
+
+import org.apache.log4j.Logger;
+
+/**
+ * This class starts and runs a standalone ZooKeeperServer.
+ */
+public class ZooKeeperServerMain {
+    
+    private static final Logger LOG = Logger.getLogger(ZooKeeperServerMain.class);
+
+    /*
+     * Start up the ZooKeeper server.
+     *
+     * @param args the port and data directory
+     */
+    public static void main(String[] args) {
+        ServerConfig.parse(args);
+        runStandalone(new ZooKeeperServer.Factory() {
+            public NIOServerCnxn.Factory createConnectionFactory() throws IOException {
+                return new NIOServerCnxn.Factory(ServerConfig.getClientPort());
+            }
+
+            public ZooKeeperServer createServer() throws IOException {
+                ZooKeeperServer zks = new ZooKeeperServer();
+                zks.setDataDir(new File(ServerConfig.getDataDir()));
+                zks.setDataLogDir(new File(ServerConfig.getDataLogDir()));
+                zks.setClientPort(ServerConfig.getClientPort());
+                return zks;
+            }
+        });
+    }
+
+    public static void runStandalone(ZooKeeperServer.Factory factory) {
+        try {
+            // Note that this thread isn't going to be doing anything else,
+            // so rather than spawning another thread, we will just call
+            // run() in this thread.
+            ServerStats.registerAsConcrete();
+            ZooKeeperServer zk = factory.createServer();
+            zk.startup();
+            NIOServerCnxn.Factory t = factory.createConnectionFactory();
+            t.setZooKeeperServer(zk);
+            t.join();
+            if (zk.isRunning()) {
+                zk.shutdown();
+            }
+        } catch (Exception e) {
+            LOG.fatal("Unexpected exception",e);
+        }
+        System.exit(0);
+    }
+}

src/java/main/org/apache/zookeeper/server/quorum/QuorumPeer.java

@@ -15,21 +15,8 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
-
 package org.apache.zookeeper.server.quorum;
 
-
-import static org.apache.zookeeper.server.ServerConfig.getClientPort;
-import static org.apache.zookeeper.server.ServerConfig.getDataDir;
-import static org.apache.zookeeper.server.ServerConfig.getDataLogDir;
-import static org.apache.zookeeper.server.quorum.QuorumPeerConfig.getElectionAlg;
-import static org.apache.zookeeper.server.quorum.QuorumPeerConfig.getElectionPort;
-import static org.apache.zookeeper.server.quorum.QuorumPeerConfig.getInitLimit;
-import static org.apache.zookeeper.server.quorum.QuorumPeerConfig.getServerId;
-import static org.apache.zookeeper.server.quorum.QuorumPeerConfig.getServers;
-import static org.apache.zookeeper.server.quorum.QuorumPeerConfig.getSyncLimit;
-import static org.apache.zookeeper.server.quorum.QuorumPeerConfig.getTickTime;
-
 import java.io.ByteArrayInputStream;
 import java.io.File;
 import java.io.FileInputStream;
@@ -42,10 +29,9 @@ import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.List;
 
-import org.apache.log4j.Logger;
-
 import org.apache.jute.BinaryInputArchive;
 import org.apache.jute.InputArchive;
+import org.apache.log4j.Logger;
 import org.apache.zookeeper.server.NIOServerCnxn;
 import org.apache.zookeeper.server.ZooKeeperServer;
 import org.apache.zookeeper.txn.TxnHeader;
@@ -76,28 +62,6 @@ import org.apache.zookeeper.txn.TxnHeader;
  * </pre>
  *
  * The request for the current leader will consist solely of an xid: int xid;
- *
- * <h2>Configuration file</h2>
- *
- * When the main() method of this class is used to start the program, the file
- * "zoo.cfg" in the current directory will be used to obtain configuration
- * information. zoo.cfg is a Properties file, so keys and values are separated
- * by equals (=) and the key/value pairs are separated by new lines. The
- * following keys are used in the configuration file:
- * <ol>
- * <li>dataDir - The directory where the zookeeper data is stored.</li>
- * <li>clientPort - The port used to communicate with clients.</li>
- * <li>tickTime - The duration of a tick in milliseconds. This is the basic
- * unit of time in zookeeper.</li>
- * <li>initLimit - The maximum number of ticks that a follower will wait to
- * initially synchronize with a leader.</li>
- * <li>syncLimit - The maximum number of ticks that a follower will wait for a
- * message (including heartbeats) from the leader.</li>
- * <li>server.<i>id</i> - This is the host:port that the server with the
- * given id will use for the quorum protocol.</li>
- * </ol>
- * In addition to the zoo.cfg file. There is a file in the data directory called
- * "myid" that contains the server id as an ASCII decimal value.
  */
 public class QuorumPeer extends Thread implements QuorumStats.Provider {
     private static final Logger LOG = Logger.getLogger(QuorumPeer.class);
@@ -267,16 +231,24 @@ public class QuorumPeer extends Thread implements QuorumStats.Provider {
      */
     private File dataLogDir;
 
+    private int electionType;
+
     Election electionAlg;
 
     int electionPort;
 
     NIOServerCnxn.Factory cnxnFactory;
 
+
+    public QuorumPeer() {
+        super("QuorumPeer");
+    }
+    
     public QuorumPeer(ArrayList<QuorumServer> quorumPeers, File dataDir,
-            File dataLogDir, int electionAlg, int electionPort,long myid, int tickTime,
+            File dataLogDir, int electionType, int electionPort,long myid, int tickTime,
             int initLimit, int syncLimit,NIOServerCnxn.Factory cnxnFactory) throws IOException {
-        super("QuorumPeer");
+        this();
+        this.electionType = electionType;
         this.cnxnFactory = cnxnFactory;
         this.quorumPeers = quorumPeers;
         this.dataDir = dataDir;
@@ -285,7 +257,14 @@ public class QuorumPeer extends Thread implements QuorumStats.Provider {
         this.myid = myid;
         this.tickTime = tickTime;
         this.initLimit = initLimit;
-        this.syncLimit = syncLimit;
+        this.syncLimit = syncLimit;        
+        
+        QuorumStats.getInstance().setStatsProvider(this);
+    }
+
+    @Override
+    public synchronized void start() {
+        
         currentVote = new Vote(myid, getLastLoggedZxid());
         for (QuorumServer p : quorumPeers) {
             if (p.id == myid) {
@@ -294,16 +273,20 @@ public class QuorumPeer extends Thread implements QuorumStats.Provider {
             }
         }
         if (myQuorumAddr == null) {
-            throw new SocketException("My id " + myid + " not in the peer list");
+            throw new RuntimeException("My id " + myid + " not in the peer list");
         }
-        if (electionAlg == 0) {
-            udpSocket = new DatagramSocket(myQuorumAddr.getPort());
-            new ResponderThread().start();
+        if (electionType == 0) {
+            try {
+                udpSocket = new DatagramSocket(myQuorumAddr.getPort());
+                new ResponderThread().start();
+            } catch (SocketException e) {
+                new RuntimeException(e);
+            }
         }
-        this.electionAlg = createElectionAlgorithm(electionAlg);
-        QuorumStats.getInstance().setStatsProvider(this);
+        this.electionAlg = createElectionAlgorithm(electionType);
+        super.start();
     }
-
+    
     /**
      * This constructor is only used by the existing unit test code.
      */
@@ -313,26 +296,23 @@ public class QuorumPeer extends Thread implements QuorumStats.Provider {
         this(quorumPeers,dataDir,dataLogDir,electionAlg,electionPort,myid,tickTime,
                 initLimit,syncLimit,new NIOServerCnxn.Factory(clientPort));
     }
-    /**
-     *  The constructor uses the quorum peer config to instantiate the class
-     */
-    public QuorumPeer(NIOServerCnxn.Factory cnxnFactory) throws IOException {
-        this(getServers(), new File(getDataDir()), new File(getDataLogDir()),
-                getElectionAlg(), getElectionPort(),getServerId(),getTickTime(),
-                getInitLimit(), getSyncLimit(),cnxnFactory);
-    }
 
     public Follower follower;
     public Leader leader;
 
+    private int clientPort;
+
     protected Follower makeFollower(File dataDir,File dataLogDir) throws IOException {
-        return new Follower(this, new FollowerZooKeeperServer(dataDir,
-                dataLogDir, this,new ZooKeeperServer.BasicDataTreeBuilder()));
+        FollowerZooKeeperServer zks = new FollowerZooKeeperServer(dataDir, dataLogDir, this,new ZooKeeperServer.BasicDataTreeBuilder());
+        zks.setClientPort(clientPort);
+        return new Follower(this, zks);
     }
 
     protected Leader makeLeader(File dataDir,File dataLogDir) throws IOException {
-        return new Leader(this, new LeaderZooKeeperServer(dataDir, dataLogDir,
-                this,new ZooKeeperServer.BasicDataTreeBuilder()));
+        LeaderZooKeeperServer zks = new LeaderZooKeeperServer(dataDir, dataLogDir,
+                this,new ZooKeeperServer.BasicDataTreeBuilder());
+        zks.setClientPort(clientPort);
+        return new Leader(this, zks);
     }
 
     private Election createElectionAlgorithm(int electionAlgorithm){
@@ -502,18 +482,6 @@ public class QuorumPeer extends Thread implements QuorumStats.Provider {
         return zxid;
     }
 
-    public static void runPeer(QuorumPeer.Factory qpFactory) {
-        try {
-            QuorumStats.registerAsConcrete();
-            QuorumPeer self = qpFactory.create(qpFactory.createConnectionFactory());
-            self.start();
-            self.join();
-        } catch (Exception e) {
-            LOG.fatal("Unexpected exception",e);
-        }
-        System.exit(2);
-    }
-
     public String[] getQuorumPeers() {
         List<String> l = new ArrayList<String>();
         synchronized (this) {
@@ -547,27 +515,138 @@ public class QuorumPeer extends Thread implements QuorumStats.Provider {
         return QuorumStats.Provider.UNKNOWN_STATE;
     }
 
-    public static void main(String args[]) {
-        if (args.length == 2) {
-            ZooKeeperServer.main(args);
-            return;
-        }
-        QuorumPeerConfig.parse(args);
+    /**
+     * get the id of this quorum peer.
+     */
+    public long getMyid() {
+        return myid;
+    }
 
-        if (!QuorumPeerConfig.isStandalone()) {
-            runPeer(new QuorumPeer.Factory() {
-                public QuorumPeer create(NIOServerCnxn.Factory cnxnFactory)
-                        throws IOException {
-                    return new QuorumPeer(cnxnFactory);
-                }
-                public NIOServerCnxn.Factory createConnectionFactory()
-                        throws IOException {
-                    return new NIOServerCnxn.Factory(getClientPort());
-                }
-            });
-        }else{
-            // there is only server in the quorum -- run as standalone
-            ZooKeeperServer.main(args);
-        }
+    /**
+     * set the id of this quorum peer.
+     */
+    public void setMyid(long myid) {
+        this.myid = myid;
+    }
+
+    /**
+     * Get the number of milliseconds of each tick
+     */
+    public int getTickTime() {
+        return tickTime;
+    }
+
+    /**
+     * Set the number of milliseconds of each tick
+     */
+    public void setTickTime(int tickTime) {
+        this.tickTime = tickTime;
+    }
+
+    /**
+     * Get the number of ticks that the initial synchronization phase can take
+     */
+    public int getInitLimit() {
+        return initLimit;
+    }
+
+    /**
+     * Set the number of ticks that the initial synchronization phase can take
+     */
+    public void setInitLimit(int initLimit) {
+        this.initLimit = initLimit;
+    }
+
+    /**
+     * Get the number of ticks that can pass between sending a request and getting
+     * an acknowledgement
+     */
+    public int getSyncLimit() {
+        return syncLimit;
+    }
+
+    /**
+     * Set the number of ticks that can pass between sending a request and getting
+     * an acknowledgement
+     */
+    public void setSyncLimit(int syncLimit) {
+        this.syncLimit = syncLimit;
+    }
+
+    /**
+     * Get the directory where the snapshot is stored.
+     */
+    public File getDataDir() {
+        return dataDir;
+    }
+
+    /**
+     * Set the directory where the snapshot is stored.
+     */
+    public void setDataDir(File dataDir) {
+        this.dataDir = dataDir;
+    }
+
+    /**
+     * Get the directory where the logs are stored.
+     */
+    public File getDataLogDir() {
+        return dataLogDir;
+    }
+
+    /**
+     * Set the directory where the logs are stored.
+     */
+    public void setDataLogDir(File dataLogDir) {
+        this.dataLogDir = dataLogDir;
     }
+
+    /**
+     * Gets the election port
+     */
+    public int getElectionPort() {
+        return electionPort;
+    }
+
+    /**
+     * Gets the election type
+     */
+    public int getElectionType() {
+        return electionType;
+    }
+
+    /**
+     * Sets the election type
+     */
+    public void setElectionType(int electionType) {
+        this.electionType = electionType;
+    }
+
+    /**
+     * Sets the election port
+     */
+    public void setElectionPort(int electionPort) {
+        this.electionPort = electionPort;
+    }
+
+    public NIOServerCnxn.Factory getCnxnFactory() {
+        return cnxnFactory;
+    }
+
+    public void setCnxnFactory(NIOServerCnxn.Factory cnxnFactory) {
+        this.cnxnFactory = cnxnFactory;
+    }
+
+    public void setQuorumPeers(ArrayList<QuorumServer> quorumPeers) {
+        this.quorumPeers = quorumPeers;
+    }
+
+    public int getClientPort() {
+        return clientPort;
+    }
+
+    public void setClientPort(int clientPort) {
+        this.clientPort = clientPort;
+    }
+
 }

src/java/main/org/apache/zookeeper/server/quorum/QuorumPeerMain.java

@@ -0,0 +1,112 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.zookeeper.server.quorum;
+
+import static org.apache.zookeeper.server.ServerConfig.getClientPort;
+
+import java.io.File;
+import java.io.IOException;
+
+import org.apache.log4j.Logger;
+import org.apache.zookeeper.server.NIOServerCnxn;
+import org.apache.zookeeper.server.ServerConfig;
+import org.apache.zookeeper.server.ZooKeeperServerMain;
+
+/**
+ * 
+ * <h2>Configuration file</h2>
+ *
+ * When the main() method of this class is used to start the program, the file
+ * "zoo.cfg" in the current directory will be used to obtain configuration
+ * information. zoo.cfg is a Properties file, so keys and values are separated
+ * by equals (=) and the key/value pairs are separated by new lines. The
+ * following keys are used in the configuration file:
+ * <ol>
+ * <li>dataDir - The directory where the zookeeper data is stored.</li>
+ * <li>clientPort - The port used to communicate with clients.</li>
+ * <li>tickTime - The duration of a tick in milliseconds. This is the basic
+ * unit of time in zookeeper.</li>
+ * <li>initLimit - The maximum number of ticks that a follower will wait to
+ * initially synchronize with a leader.</li>
+ * <li>syncLimit - The maximum number of ticks that a follower will wait for a
+ * message (including heartbeats) from the leader.</li>
+ * <li>server.<i>id</i> - This is the host:port that the server with the
+ * given id will use for the quorum protocol.</li>
+ * </ol>
+ * In addition to the zoo.cfg file. There is a file in the data directory called
+ * "myid" that contains the server id as an ASCII decimal value.
+ * 
+ */
+public class QuorumPeerMain {
+    
+    private static final Logger LOG = Logger.getLogger(QuorumPeerMain.class);
+
+    /**
+     * To start the replicated server specify the configuration file name on the
+     * command line.
+     * @param args command line
+     */
+    public static void main(String[] args) {
+        if (args.length == 2) {
+            ZooKeeperServerMain.main(args);
+            return;
+        }
+        QuorumPeerConfig.parse(args);
+        if (!QuorumPeerConfig.isStandalone()) {
+            runPeer(new QuorumPeer.Factory() {
+                public QuorumPeer create(NIOServerCnxn.Factory cnxnFactory)
+                        throws IOException {
+                    
+                    QuorumPeer peer = new QuorumPeer();
+                    peer.setClientPort(ServerConfig.getClientPort());
+                    peer.setDataDir(new File(ServerConfig.getDataDir()));
+                    peer.setDataLogDir(new File(ServerConfig.getDataLogDir()));
+                    peer.setQuorumPeers(QuorumPeerConfig.getServers());
+                    peer.setElectionPort(QuorumPeerConfig.getElectionPort());
+                    peer.setElectionType(QuorumPeerConfig.getElectionAlg());
+                    peer.setMyid(QuorumPeerConfig.getServerId());
+                    peer.setTickTime(QuorumPeerConfig.getTickTime());
+                    peer.setInitLimit(QuorumPeerConfig.getInitLimit());
+                    peer.setSyncLimit(QuorumPeerConfig.getSyncLimit());
+                    peer.setCnxnFactory(cnxnFactory);
+                    return peer;
+                    
+                }
+                public NIOServerCnxn.Factory createConnectionFactory() throws IOException {
+                    return new NIOServerCnxn.Factory(getClientPort());
+                }
+            });
+        }else{
+            // there is only server in the quorum -- run as standalone
+            ZooKeeperServerMain.main(args);
+        }
+    }
+    
+    public static void runPeer(QuorumPeer.Factory qpFactory) {
+        try {
+            QuorumStats.registerAsConcrete();
+            QuorumPeer self = qpFactory.create(qpFactory.createConnectionFactory());
+            self.start();
+            self.join();
+        } catch (Exception e) {
+            LOG.fatal("Unexpected exception",e);
+        }
+        System.exit(2);
+    }
+
+}