ZOOKEEPER-136.patch

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

src/java/main/org/apache/jute/CsvOutputArchive.java

@@ -103,6 +103,9 @@ public class CsvOutputArchive implements OutputArchive {
     }
     
     public void writeRecord(Record r, String tag) throws IOException {
+        if (r == null) {
+            return;
+        }
         r.serialize(this, tag);
     }
     

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

@@ -73,8 +73,7 @@ import org.apache.zookeeper.server.ZooTrace;
 public class ClientCnxn {
     private static final Logger LOG = Logger.getLogger(ClientCnxn.class);
 
-    private ArrayList<InetSocketAddress> serverAddrs =
-        new ArrayList<InetSocketAddress>();
+    private ArrayList<InetSocketAddress> serverAddrs = new ArrayList<InetSocketAddress>();
 
     static class AuthData {
         AuthData(String scheme, byte data[]) {
@@ -121,10 +120,12 @@ public class ClientCnxn {
 
     final Selector selector = Selector.open();
     
-    /** Set to true when close is called. Latches the connection such that
-     * we don't attempt to re-connect to the server if in the middle of
-     * closing the connection (client sends session disconnect to server
-     * as part of close operation) */
+    /**
+     * Set to true when close is called. Latches the connection such that we
+     * don't attempt to re-connect to the server if in the middle of closing the
+     * connection (client sends session disconnect to server as part of close
+     * operation)
+     */
     volatile boolean closing = false;
 
     public long getSessionId() {
@@ -138,7 +139,8 @@ public class ClientCnxn {
     @Override
     public String toString() {
         StringBuffer sb = new StringBuffer();
-        sb.append("sessionId: 0x").append(Long.toHexString(getSessionId())).append("\n");
+        sb.append("sessionId: 0x").append(Long.toHexString(getSessionId()))
+          .append("\n");
         sb.append("lastZxid: ").append(lastZxid).append("\n");
         sb.append("xid: ").append(xid).append("\n");
         sb.append("nextAddrToTry: ").append(nextAddrToTry).append("\n");
@@ -200,8 +202,25 @@ public class ClientCnxn {
             }
             this.watchRegistration = watchRegistration;
         }
+
+        @Override
+        public String toString() {
+            StringBuffer sb = new StringBuffer();
+
+            sb.append("path:" + path);
+            sb.append(" finished:" + finished);
+
+            sb.append(" header:: " + header);
+            sb.append(" replyHeader:: " + replyHeader);
+            sb.append(" request:: " + request);
+            sb.append(" response:: " + response);
+
+            // jute toString is horrible, remove unnecessary newlines
+            return sb.toString().replaceAll("\r*\n+", " ");
+        }
     }
 
+
     public ClientCnxn(String hosts, int sessionTimeout, ZooKeeper zooKeeper,
             ClientWatchManager watcher)
         throws IOException
@@ -224,8 +243,7 @@ public class ClientCnxn {
      */
     public ClientCnxn(String hosts, int sessionTimeout, ZooKeeper zooKeeper,
             ClientWatchManager watcher, long sessionId, byte[] sessionPasswd)
-        throws IOException
-    {
+        throws IOException {
         this.zooKeeper = zooKeeper;
         this.watcher = watcher;
         this.sessionId = sessionId;
@@ -477,20 +495,28 @@ public class ClientCnxn {
             ByteBufferInputStream bbis = new ByteBufferInputStream(
                     incomingBuffer);
             BinaryInputArchive bbia = BinaryInputArchive.getArchive(bbis);
-            ReplyHeader r = new ReplyHeader();
+            ReplyHeader replyHdr = new ReplyHeader();
 
-            r.deserialize(bbia, "header");
-            if (r.getXid() == -2) {
+            replyHdr.deserialize(bbia, "header");
+            if (replyHdr.getXid() == -2) {
                 // -2 is the xid for pings
+                LOG
+                    .debug("Got ping sessionid:0x"
+                    + Long.toHexString(sessionId));
                 return;
             }
-            if (r.getXid() == -4) {
+            if (replyHdr.getXid() == -4) {
                 // -2 is the xid for AuthPacket
                 // TODO: process AuthPacket here
+                LOG
+                    .debug("Got auth sessionid:0x"
+                        + Long.toHexString(sessionId));
                 return;
             }
-            if (r.getXid() == -1) {
+            if (replyHdr.getXid() == -1) {
                 // -1 means notification
+                LOG.debug("Got notification sessionid:0x"
+                    + Long.toHexString(sessionId));
                 WatcherEvent event = new WatcherEvent();
                 event.deserialize(bbia, "response");
                 WatchedEvent we = new WatchedEvent(event);
@@ -504,29 +530,37 @@ public class ClientCnxn {
             }
             if (pendingQueue.size() == 0) {
                 throw new IOException("Nothing in the queue, but got "
-                        + r.getXid());
+                        + replyHdr.getXid());
             }
-            Packet p = null;
+            Packet packet = null;
             synchronized (pendingQueue) {
-                p = pendingQueue.remove();
+                packet = pendingQueue.remove();
             }
             /*
              * Since requests are processed in order, we better get a response
              * to the first request!
              */
-            if (p.header.getXid() != r.getXid()) {
-                throw new IOException("Xid out of order. Got " + r.getXid()
-                        + " expected " + p.header.getXid());
+         if (packet.header.getXid() != replyHdr.getXid()) {
+         throw new IOException("Xid out of order. Got "
+                  + replyHdr.getXid() + " expected "
+                  + packet.header.getXid());
             }
-            p.replyHeader.setXid(r.getXid());
-            p.replyHeader.setErr(r.getErr());
-            p.replyHeader.setZxid(r.getZxid());
-            lastZxid = r.getZxid();
-            if (p.response != null && r.getErr() == 0) {
-                p.response.deserialize(bbia, "response");
+
+         packet.replyHeader.setXid(replyHdr.getXid());
+         packet.replyHeader.setErr(replyHdr.getErr());
+         packet.replyHeader.setZxid(replyHdr.getZxid());
+         lastZxid = replyHdr.getZxid();
+         if (packet.response != null && replyHdr.getErr() == 0) {
+             packet.response.deserialize(bbia, "response");
             }
-            p.finished = true;
-            finishPacket(p);
+            packet.finished = true;
+
+            if (LOG.isDebugEnabled()) {
+            LOG.debug("Reading reply sessionid:0x"
+                + Long.toHexString(sessionId) + ", packet:: " + packet);
+        }
+
+            finishPacket(packet);
         }
 
         /**
@@ -789,14 +823,15 @@ public class ClientCnxn {
                 } catch (Exception e) {
                     if (closing) {
                         // closing so this is expected
-                        LOG.info("Exception while closing send thread for session 0x" 
+                        LOG
+                           .info("Exception while closing send thread for session 0x"
                                 + Long.toHexString(getSessionId())
                                 + " : " + e.getMessage());
                         break;
                     } else {
                         LOG.warn("Exception closing session 0x" 
-                                + Long.toHexString(getSessionId()),
-                                e);
+                                + Long.toHexString(getSessionId()) + " to "
+                                + sockKey, e);
                         cleanup();
                         if (zooKeeper.state.isAlive()) {
                             eventThread.queueEvent(new WatchedEvent(
@@ -889,8 +924,8 @@ public class ClientCnxn {
     }
 
     /**
-     * Close the connection, which includes; send session disconnect to
-     * the server, shutdown the send/event threads.
+     * Close the connection, which includes; send session disconnect to the
+     * server, shutdown the send/event threads.
      * 
      * @throws IOException
      */
@@ -919,13 +954,10 @@ public class ClientCnxn {
     }
 
     public ReplyHeader submitRequest(RequestHeader h, Record request,
-            Record response,
-            WatchRegistration watchRegistration)
-        throws InterruptedException
-    {
+            Record response, WatchRegistration watchRegistration)
+            throws InterruptedException {
         ReplyHeader r = new ReplyHeader();
-        Packet packet =
-            queuePacket(h, r, request, response, null, null, null,
+        Packet packet = queuePacket(h, r, request, response, null, null, null,
                     watchRegistration);
         synchronized (packet) {
             while (!packet.finished) {
@@ -937,8 +969,7 @@ public class ClientCnxn {
 
     Packet queuePacket(RequestHeader h, ReplyHeader r, Record request,
             Record response, AsyncCallback cb, String path, Object ctx,
-            WatchRegistration watchRegistration)
-    {
+            WatchRegistration watchRegistration) {
         Packet packet = null;
         synchronized (outgoingQueue) {
             if (h.getType() != OpCode.ping && h.getType() != OpCode.auth) {

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

@@ -69,9 +69,9 @@ public class FinalRequestProcessor implements RequestProcessor {
     }
 
     public void processRequest(Request request) {
-        // LOG.info("Zoo>>> cxid = " + request.cxid + " type = " +
-        // request.type + " id = " + request.sessionId + " cnxn " +
-        // request.cnxn);
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("Processing request:: " + request);
+        }
         // request.addRQRec(">final");
         long traceMask = ZooTrace.CLIENT_REQUEST_TRACE_MASK;
         if (request.type == OpCode.ping) {
@@ -130,6 +130,10 @@ public class FinalRequestProcessor implements RequestProcessor {
             if (request.hdr != null && request.hdr.getType() == OpCode.error) {
                 throw KeeperException.create(((ErrorTxn) request.txn).getErr());
             }
+
+            if (LOG.isDebugEnabled()) {
+                LOG.debug(request);
+            }
             switch (request.type) {
             case OpCode.ping:
                 request.cnxn.sendResponse(new ReplyHeader(-2,
@@ -157,7 +161,6 @@ public class FinalRequestProcessor implements RequestProcessor {
                 err = rc.err;
                 break;
             case OpCode.sync:
-                LOG.debug("OpCode.sync " + request);
                 SyncRequest syncRequest = new SyncRequest();
                 ZooKeeperServer.byteBuffer2Record(request.request,
                         syncRequest);

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

@@ -84,7 +84,7 @@ public class NIOServerCnxn implements Watcher, ServerCnxn {
         int outstandingLimit = 1;
 
         public Factory(int port) throws IOException {
-            super("NIOServerCxn.Factory");
+            super("NIOServerCxn.Factory:" + port);
             setDaemon(true);
             this.ss = ServerSocketChannel.open();
             ss.socket().bind(new InetSocketAddress(port));
@@ -421,7 +421,7 @@ public class NIOServerCnxn implements Watcher, ServerCnxn {
                     || ap.handleAuthentication(this, authPacket.getAuth()) != KeeperException.Code.Ok) {
                 if (ap == null)
                     LOG.error("No authentication provider for scheme: "
-                            + scheme);
+                            + scheme + " has " + ProviderRegistry.listProviders());
                 else
                     LOG.debug("Authentication failed for scheme: "
                             + scheme);
@@ -449,6 +449,7 @@ public class NIOServerCnxn implements Watcher, ServerCnxn {
                 outstandingRequests++;
                 // check throttling
                 if (zk.getInProcess() > factory.outstandingLimit) {
+                    LOG.warn("Throttling recv " + zk.getInProcess());
                     disableRecv();
                     // following lines should not be needed since we are already
                     // reading

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

@@ -76,7 +76,7 @@ public class PrepRequestProcessor extends Thread implements RequestProcessor {
 
     public PrepRequestProcessor(ZooKeeperServer zks,
             RequestProcessor nextProcessor) {
-        super("ProcessThread");
+        super("ProcessThread:" + zks.getClientPort());
         this.nextProcessor = nextProcessor;
         this.zks = zks;
 

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

@@ -163,19 +163,19 @@ public class Request {
     @Override
     public String toString() {
         StringBuffer sb = new StringBuffer();
-        sb.append("session 0x").append(Long.toHexString(sessionId));
-        sb.append(" cxid 0x").append(Long.toHexString(cxid));
-        sb.append("zxid 0x").append(Long.toHexString((hdr == null ? -2 : hdr.getZxid()))).append(
-                " ");
-        sb
-                .append(
-                        " txn type = "
-                                + (hdr == null ? "unknown" : "" + hdr.getType()))
-                .append(" ");
-        sb.append(op2String(type)).append(" ");
+        sb.append("sessionid:0x").append(Long.toHexString(sessionId));
+        sb.append(" type:").append(op2String(type));
+        sb.append(" cxid:0x").append(Long.toHexString(cxid));
+        sb.append(" zxid:0x").append(Long.toHexString((hdr == null ?
+                -2 : hdr.getZxid())));
+        sb.append(" txntype:" + (hdr == null ?
+                "unknown" : "" + hdr.getType()));
+        sb.append(" ");
 
         String path = "n/a";
-        if (type != OpCode.createSession) {
+        if (type != OpCode.createSession && request != null
+                && request.remaining() >= 4)
+        {
             try {
                 request.clear();
                 int pathLen = request.getInt();
@@ -188,6 +188,7 @@ public class Request {
             }
         }
         sb.append(path).append(" ");
+
         return sb.toString();
     }
 }

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

@@ -54,7 +54,7 @@ public class SyncRequestProcessor extends Thread implements RequestProcessor {
 
     public SyncRequestProcessor(ZooKeeperServer zks,
             RequestProcessor nextProcessor) {
-        super("SyncThread");
+        super("SyncThread:" + zks.getClientPort());
         this.zks = zks;
         this.nextProcessor = nextProcessor;
         start();

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

@@ -467,6 +467,11 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
      */
     public void submitRequest(ServerCnxn cnxn, long sessionId, int type,
             int xid, ByteBuffer bb, List<Id> authInfo) {
+        Request si = new Request(cnxn, sessionId, xid, type, bb, authInfo);
+        submitRequest(si);
+    }
+    
+    public void submitRequest(Request si) {
         if (firstProcessor == null) {
             synchronized (this) {
                 try {
@@ -482,16 +487,15 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
             }
         }
         try {
-            touch(cnxn);
-            Request si = new Request(cnxn, sessionId, xid, type, bb, authInfo);
-            boolean validpacket = Request.isValid(type);
+            touch(si.cnxn);
+            boolean validpacket = Request.isValid(si.type);
             if (validpacket) {
                 firstProcessor.processRequest(si);
-                if (cnxn != null) {
+                if (si.cnxn != null) {
                     incInProcess();
                 }
             } else {
-                LOG.warn("Dropping packet at server of type " + type);
+                LOG.warn("Dropping packet at server of type " + si.type);
                 // if unvalid packet drop the packet.
             }
         } catch (IOException e) {

src/java/main/org/apache/zookeeper/server/auth/ProviderRegistry.java

@@ -36,7 +36,6 @@ public class ProviderRegistry {
         synchronized (ProviderRegistry.class) {
             if (initialized)
                 return;
-            initialized = true;
             IPAuthenticationProvider ipp = new IPAuthenticationProvider();
             HostAuthenticationProvider hostp = new HostAuthenticationProvider();
             DigestAuthenticationProvider digp = new DigestAuthenticationProvider();
@@ -59,6 +58,7 @@ public class ProviderRegistry {
                     }
                 }
             }
+            initialized = true;
         }
     }
 
@@ -67,4 +67,12 @@ public class ProviderRegistry {
             initialize();
         return authenticationProviders.get(scheme);
     }
+
+    public static String listProviders() {
+        StringBuilder sb = new StringBuilder();
+        for(String s: authenticationProviders.keySet()) {
+        sb.append(s + " ");
+}
+        return sb.toString();
+    }
 }

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

@@ -49,8 +49,17 @@ public class CommitProcessor extends Thread implements RequestProcessor {
 
     RequestProcessor nextProcessor;
 
-    public CommitProcessor(RequestProcessor nextProcessor) {
+    /**
+     * This flag indicates whether we need to wait for a response to come back from the
+     * leader or we just let the sync operation flow through like a read. The flag will
+     * be true if the CommitProcessor is in a Leader pipeline.
+     */
+    boolean matchSyncs;
+
+    public CommitProcessor(RequestProcessor nextProcessor, String id, boolean matchSyncs) {
+        super("CommitProcessor:" + id);
         this.nextProcessor = nextProcessor;
+        this.matchSyncs = matchSyncs;
         start();
     }
 
@@ -122,8 +131,11 @@ public class CommitProcessor extends Thread implements RequestProcessor {
                             nextPending = request;
                             break;
                         case OpCode.sync:
-                            nextPending = request;
-                            //pendingSyncs.add(request);
+                            if (matchSyncs) {
+                                nextPending = request;
+                            } else {
+                                toProcess.add(request);
+                            }
                             break;
                         default:
                             toProcess.add(request);
@@ -145,7 +157,9 @@ public class CommitProcessor extends Thread implements RequestProcessor {
                          new Exception("committing a null! "));
                 return;
             }
-            LOG.debug("Committing" + request.cxid);
+            if (LOG.isDebugEnabled()) {
+                LOG.debug("Committing request:: " + request);
+            }
             committedRequests.add(request);
             notifyAll();
         }
@@ -153,9 +167,10 @@ public class CommitProcessor extends Thread implements RequestProcessor {
 
     synchronized public void processRequest(Request request) {
         // request.addRQRec(">commit");
-        // LOG.info("Zoo processReq>>> cxid = " + request.cxid + " type =
-        // " + request.type + " id = " + request.sessionId + " cnxn " +
-        // request.cnxn);
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("Processing request:: " + request);
+        }
+        
         if (!finished) {
             queuedRequests.add(request);
             notifyAll();

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

@@ -330,10 +330,12 @@ public class FollowerHandler extends Thread {
                     type = bb.getInt();
                     bb = bb.slice();
                     if(type == OpCode.sync){
-                        leader.setSyncHandler(this, sessionId);
-                    }
+                     	leader.zk.submitRequest(new FollowerSyncRequest(this, sessionId, cxid, type, bb,
+                                qp.getAuthinfo()));
+                    } else {
                     leader.zk.submitRequest(null, sessionId, type, cxid, bb,
                             qp.getAuthinfo());
+                    }
                     break;
                 default:
                 }

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

@@ -45,6 +45,7 @@ public class FollowerRequestProcessor extends Thread implements
 
     public FollowerRequestProcessor(FollowerZooKeeperServer zks,
             RequestProcessor nextProcessor) {
+        super("FollowerRequestProcessor:" + zks.getClientPort());
         this.zks = zks;
         this.nextProcessor = nextProcessor;
         start();

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

@@ -0,0 +1,16 @@
+package org.apache.zookeeper.server.quorum;
+
+import java.nio.ByteBuffer;
+import java.util.List;
+
+import org.apache.zookeeper.data.Id;
+import org.apache.zookeeper.server.Request;
+
+public class FollowerSyncRequest extends Request {
+	FollowerHandler fh;
+	public FollowerSyncRequest(FollowerHandler fh, long sessionId, int xid, int type,
+			ByteBuffer bb, List<Id> authInfo) {
+		super(null, sessionId, xid, type, bb, authInfo);
+		this.fh = fh;
+	}
+}

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

@@ -80,7 +80,8 @@ public class FollowerZooKeeperServer extends ZooKeeperServer {
     @Override
     protected void setupRequestProcessors() {
         RequestProcessor finalProcessor = new FinalRequestProcessor(this);
-        commitProcessor = new CommitProcessor(finalProcessor);
+        commitProcessor = new CommitProcessor(finalProcessor,
+                Integer.toString(getClientPort()), true);
         firstProcessor = new FollowerRequestProcessor(this, commitProcessor);
         syncProcessor = new SyncRequestProcessor(this,
                 new SendAckRequestProcessor(getFollower()));
@@ -135,16 +136,16 @@ public class FollowerZooKeeperServer extends ZooKeeperServer {
         commitProcessor.commit(request);
     }
     
-    public void sync(){
+    synchronized public void sync(){
         if(pendingSyncs.size() ==0){
             LOG.warn("Not expecting a sync.");
             return;
         }
                 
-        commitProcessor.commit(pendingSyncs.remove());
+        Request r = pendingSyncs.remove();
+		commitProcessor.commit(r);
     }
              
-         
     @Override
     public int getGlobalOutstandingLimit() {
         return super.getGlobalOutstandingLimit() / (self.getQuorumSize() - 1);

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

@@ -25,9 +25,11 @@ import java.net.ServerSocket;
 import java.net.Socket;
 import java.net.SocketAddress;
 import java.net.SocketException;
+import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Iterator;
+import java.util.List;
 import java.util.concurrent.ConcurrentLinkedQueue;
 
 import org.apache.log4j.Logger;
@@ -70,11 +72,8 @@ public class Leader {
     public HashSet<FollowerHandler> forwardingFollowers = new HashSet<FollowerHandler>();
     
     //Pending sync requests
-    public HashMap<Long,Request> pendingSyncs = new HashMap<Long,Request>();
+    public HashMap<Long,List<FollowerSyncRequest>> pendingSyncs = new HashMap<Long,List<FollowerSyncRequest>>();
                
-    //Map sync request to FollowerHandler
-    public HashMap<Long,FollowerHandler> syncHandler = new HashMap<Long,FollowerHandler>();
-       
     /**
      * Adds follower to the leader.
      * 
@@ -253,7 +252,7 @@ public class Leader {
         newLeaderProposal.packet = new QuorumPacket(NEWLEADER, zk.getZxid(),
                 null, null);
         if ((newLeaderProposal.packet.getZxid() & 0xffffffffL) != 0) {
-            LOG.error("NEWLEADER proposal has Zxid of "
+            LOG.warn("NEWLEADER proposal has Zxid of "
                     + newLeaderProposal.packet.getZxid());
         }
         outstandingProposals.add(newLeaderProposal);
@@ -373,17 +372,29 @@ public class Leader {
      */
     synchronized public void processAck(long zxid, SocketAddress followerAddr) {
         boolean first = true;
-        /*
-         * LOG.error("Ack zxid: " + Long.toHexString(zxid)); for (Proposal
-         * p : outstandingProposals) { long packetZxid = p.packet.getZxid();
-         * LOG.error("outstanding proposal: " +
-         * Long.toHexString(packetZxid)); } LOG.error("outstanding
-         * proposals all");
-         */
+        
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("Ack zxid: 0x" + Long.toHexString(zxid));
+            for (Proposal p : outstandingProposals) {
+                long packetZxid = p.packet.getZxid();
+                LOG.debug("outstanding proposal: 0x"
+                        + Long.toHexString(packetZxid));
+            }
+            LOG.debug("outstanding proposals all");
+        }
+        
         if (outstandingProposals.size() == 0) {
+            if (LOG.isDebugEnabled()) {
+                LOG.debug("outstanding is 0");
+            }
             return;
         }
         if (outstandingProposals.peek().packet.getZxid() > zxid) {
+            if (LOG.isDebugEnabled()) {
+                LOG.debug("proposal has already been committed, pzxid:"
+                        + outstandingProposals.peek().packet.getZxid()
+                        + " zxid:" + zxid);
+            }
             // The proposal has already been committed
             return;
         }
@@ -391,13 +402,16 @@ public class Leader {
             long packetZxid = p.packet.getZxid();
             if (packetZxid == zxid) {
                 p.ackCount++;
-                // LOG.error("FIXMSG",new RuntimeException(), "Count for " +
-                // Long.toHexString(zxid) + " is " + p.ackCount);
+                if (LOG.isDebugEnabled()) {
+                    LOG.debug("Count for zxid: 0x" + Long.toHexString(zxid)
+                            + " is " + p.ackCount);
+                }
+                
                 if (p.ackCount > self.quorumPeers.size() / 2){
                     if (!first) {
-                        LOG.error("Commiting zxid 0x" + Long.toHexString(zxid)
+                        LOG.fatal("Commiting zxid 0x" + Long.toHexString(zxid)
                                 + " from " + followerAddr + " not first!");
-                        LOG.error("First is "
+                        LOG.fatal("First is "
                                 + outstandingProposals.element().packet);
                         System.exit(13);
                     }
@@ -408,23 +422,23 @@ public class Leader {
                     // We don't commit the new leader proposal
                     if ((zxid & 0xffffffffL) != 0) {
                         if (p.request == null) {
-                            LOG.error("Going to commmit null: " + p);
+                            LOG.warn("Going to commmit null: " + p);
                         }
                         commit(zxid);
                         zk.commitProcessor.commit(p.request);
                         if(pendingSyncs.containsKey(zxid)){
-                            sendSync(syncHandler.get(pendingSyncs.get(zxid).sessionId), pendingSyncs.get(zxid));
-                            syncHandler.remove(pendingSyncs.get(zxid));
-                            pendingSyncs.remove(zxid);
+                           for(FollowerSyncRequest r: pendingSyncs.remove(zxid)) {
+                               sendSync(r);
                         }
                     }
                 }
+                }
                 return;
             } else {
                 first = false;
             }
         }
-        LOG.error("Trying to commit future proposal: zxid 0x"
+        LOG.warn("Trying to commit future proposal: zxid 0x"
                 + Long.toHexString(zxid) + " from " + followerAddr);
     }
 
@@ -528,8 +542,7 @@ public class Leader {
             }
             baos.close();
         } catch (IOException e) {
-            // This really should be impossible
-            LOG.error("FIXMSG",e);
+            LOG.warn("This really should be impossible", e);
         }
         QuorumPacket pp = new QuorumPacket(Leader.PROPOSAL, request.zxid, baos
                 .toByteArray(), null);
@@ -538,6 +551,10 @@ public class Leader {
         p.packet = pp;
         p.request = request;
         synchronized (this) {
+            if (LOG.isDebugEnabled()) {
+                LOG.debug("Proposing:: " + request);
+            }
+
             outstandingProposals.add(p);
             lastProposed = p.packet.getZxid();
             sendPacket(pp);
@@ -551,28 +568,19 @@ public class Leader {
      * @param r the request
      */
     
-    public void processSync(Request r){
+    synchronized public void processSync(FollowerSyncRequest r){
         if(outstandingProposals.isEmpty()){
-            LOG.warn("No outstanding proposal");
-            sendSync(syncHandler.get(r.sessionId), r);
-                syncHandler.remove(r.sessionId);
-        }
-        else{
-            pendingSyncs.put(lastProposed, r);
+            sendSync(r);
+        } else {
+            List<FollowerSyncRequest> l = pendingSyncs.get(lastProposed);
+            if (l == null) {
+                l = new ArrayList<FollowerSyncRequest>();
+            }
+            l.add(r);
+            pendingSyncs.put(lastProposed, l);
         }
     }
         
-    /**
-     * Set FollowerHandler for sync.
-     * 
-     * @param f
-     * @param s
-     */
-        
-    synchronized public void setSyncHandler(FollowerHandler f, long s){
-        syncHandler.put(s, f);
-    }
-            
     /**
      * Sends a sync message to the appropriate server
      * 
@@ -580,15 +588,9 @@ public class Leader {
      * @param r
      */
             
-    public void sendSync(FollowerHandler f, Request r){
-        if(f != null){
-            QuorumPacket qp = new QuorumPacket(Leader.SYNC, 0, null, null);
-            f.queuePacket(qp);
-        }
-        else{
-            LOG.warn("Committing sync: " + r.cxid );
-            zk.commitProcessor.commit(r);
-        }
+    public void sendSync(FollowerSyncRequest r){
+        QuorumPacket qp = new QuorumPacket(Leader.SYNC, 0, null, null);
+        r.fh.queuePacket(qp);
     }
                 
     /**

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

@@ -59,7 +59,8 @@ public class LeaderZooKeeperServer extends ZooKeeperServer {
         RequestProcessor finalProcessor = new FinalRequestProcessor(this);
         RequestProcessor toBeAppliedProcessor = new Leader.ToBeAppliedRequestProcessor(
                 finalProcessor, getLeader().toBeApplied);
-        commitProcessor = new CommitProcessor(toBeAppliedProcessor);
+        commitProcessor = new CommitProcessor(toBeAppliedProcessor,
+                Integer.toString(getClientPort()), false);
         RequestProcessor proposalProcessor = new ProposalRequestProcessor(this,
                 commitProcessor);
         firstProcessor = new PrepRequestProcessor(this, proposalProcessor);

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

@@ -56,17 +56,10 @@ public class ProposalRequestProcessor implements RequestProcessor {
          * call processRequest on the next processor.
          */
         
-        if(request.type == ZooDefs.OpCode.sync){
-            zks.getLeader().processSync(request);
-
-            if(!zks.getLeader().syncHandler.containsKey(request.sessionId)){
-                zks.getLeader().syncHandler.put(request.sessionId, null);
+        if(request instanceof FollowerSyncRequest){
+            zks.getLeader().processSync((FollowerSyncRequest)request);
+        } else {
                 nextProcessor.processRequest(request);
-            }
-            
-        }
-        else{
-            nextProcessor.processRequest(request);
             if (request.hdr != null) {
                 // We need to sync and get consensus on any transactions
                 zks.getLeader().propose(request);

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

@@ -384,6 +384,8 @@ public class QuorumPeer extends Thread implements QuorumStats.Provider {
 
     @Override
     public void run() {
+        setName("QuorumPeer:" + cnxnFactory.getLocalAddress());
+
         /*
          * Main loop
          */

src/java/test/org/apache/zookeeper/test/AsyncOps.java

@@ -123,7 +123,7 @@ public class AsyncOps {
                 fail("unexpected interrupt");
             }
             // on the lookout for timeout
-            assertSame(latch.getCount(), 0L);
+            assertSame(0L, latch.getCount());
             
             String actual = toString();
             
@@ -410,11 +410,15 @@ public class AsyncOps {
             verify();
         }
         
+        public void setData() {
+            zk.setData(path, data, version, this, toString());
+        }
+        
         public void verifySetData() {
             stat.setVersion(1);
             new StringCB(zk).verifyCreate();
 
-            zk.setData(path, data, version, this, toString());
+            setData();
             verify();
         }
         
@@ -460,10 +464,14 @@ public class AsyncOps {
             super(zk, latch);
         }
         
+        public void delete() {
+            zk.delete(path, version, this, toString());
+        }
+        
         public void verifyDelete() {
             new StringCB(zk).verifyCreate();
 
-            zk.delete(path, version, this, toString());
+            delete();
             verify();
         }
         
@@ -473,8 +481,12 @@ public class AsyncOps {
             verify();
         }
         
-        public void verifySync() {
+        public void sync() {
             zk.sync(path, this, toString());
+        }
+        
+        public void verifySync() {
+            sync();
             verify();
         }
         

src/java/test/org/apache/zookeeper/test/AsyncOpsTest.java

@@ -133,12 +133,12 @@ public class AsyncOpsTest extends ClientBase {
 
     @Test
     public void testAsyncExists() {
-        new StatCB(zk).verifySetData();
+        new StatCB(zk).verifyExists();
     }
 
     @Test
     public void testAsyncExistsFailure_NoNode() {
-        new StatCB(zk).verifySetData();
+        new StatCB(zk).verifyExistsFailure_NoNode();
     }
 
     @Test