ZOOKEEPER-417. stray message problem when changing servers (breed via mahadev)

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

CHANGES.txt

@@ -133,6 +133,9 @@ connected (breed via mahadev)
 
   ZOOKEEPER-448. png files do nto work with forrest. (mahadev)
 
+  ZOOKEEPER-417. stray message problem when changing servers (breed via
+mahadev)
+
 IMPROVEMENTS:
   ZOOKEEPER-308. improve the atomic broadcast performance 3x.
   (breed via mahadev)

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

@@ -118,7 +118,9 @@ public abstract class KeeperException extends Exception {
                 return new SessionExpiredException();
             case INVALIDCALLBACK:
                 return new InvalidCallbackException();
-
+            case SESSIONMOVED:
+                return new SessionMovedException();
+            	
             case OK:
             default:
                 throw new IllegalArgumentException("Invalid exception code");
@@ -264,6 +266,10 @@ public abstract class KeeperException extends Exception {
          */
         @Deprecated
         public static final int AuthFailed = -115;
+        /**
+         * This value will be used directly in {@link CODE#SESSIONMOVED}
+         */
+        // public static final int SessionMoved = -116;
     }
 
     /** Codes which represent the various KeeperException
@@ -296,7 +302,7 @@ public abstract class KeeperException extends Exception {
         OPERATIONTIMEOUT (OperationTimeout),
         /** Invalid arguments */
         BADARGUMENTS (BadArguments),
-
+        
         /** API errors.
          * This is never thrown by the server, it shouldn't be used other than
          * to indicate a range. Specifically error codes greater than this
@@ -324,7 +330,9 @@ public abstract class KeeperException extends Exception {
         /** Invalid ACL specified */
         INVALIDACL (InvalidACL),
         /** Client authentication failed */
-        AUTHFAILED (AuthFailed);
+        AUTHFAILED (AuthFailed),
+        /** Session moved to another server, so operation is ignored */
+        SESSIONMOVED (-116);
 
         private static final Map<Integer,Code> lookup
             = new HashMap<Integer,Code>();
@@ -397,6 +405,8 @@ public abstract class KeeperException extends Exception {
                 return "Session expired";
             case INVALIDCALLBACK:
                 return "Invalid callback";
+            case SESSIONMOVED:
+                return "Session moved";
             default:
                 return "Unknown error " + code;
         }
@@ -601,6 +611,15 @@ public abstract class KeeperException extends Exception {
             super(Code.SESSIONEXPIRED);
         }
     }
+    
+    /**
+     * @see Code#SESSIONMOVED
+     */
+    public static class SessionMovedException extends KeeperException {
+        public SessionMovedException() {
+            super(Code.SESSIONMOVED);
+        }
+    }
 
     /**
      * @see Code#SYSTEMERROR

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

@@ -137,6 +137,11 @@ public class FinalRequestProcessor implements RequestProcessor {
                 throw KeeperException.create(KeeperException.Code.get((
                         (ErrorTxn) request.txn).getErr()));
             }
+            
+            KeeperException ke = request.getException();
+            if (ke != null) {
+                throw ke;
+            }
 
             if (LOG.isDebugEnabled()) {
                 LOG.debug(request);

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

@@ -70,6 +70,10 @@ public class NIOServerCnxn implements Watcher, ServerCnxn {
 
     private ConnectionBean jmxConnectionBean;
 
+    // This is just an arbitrary object to represent requests issued by
+    // (aka owned by) this class
+    final private static Object me = new Object();
+
     static public class Factory extends Thread {
         ZooKeeperServer zks;
 
@@ -280,7 +284,12 @@ public class NIOServerCnxn implements Watcher, ServerCnxn {
 
         synchronized void closeSession(long sessionId) {
             selector.wakeup();
-            synchronized (cnxns) {
+            closeSessionWithoutWakeup(sessionId);
+        }
+
+
+		private void closeSessionWithoutWakeup(long sessionId) {
+			synchronized (cnxns) {
                 for (Iterator<NIOServerCnxn> it = cnxns.iterator(); it
                         .hasNext();) {
                     NIOServerCnxn cnxn = it.next();
@@ -534,8 +543,9 @@ public class NIOServerCnxn implements Watcher, ServerCnxn {
             }
             return;
         } else {
-            zk.submitRequest(this, sessionId, h.getType(), h.getXid(),
-                    incomingBuffer, authInfo);
+            Request si = new Request(this, sessionId, h.getXid(), h.getType(), incomingBuffer, authInfo);
+            si.setOwner(me);
+            zk.submitRequest(si);
         }
         if (h.getXid() >= 0) {
             synchronized (this) {
@@ -599,6 +609,7 @@ public class NIOServerCnxn implements Watcher, ServerCnxn {
         // session is setup
         disableRecv();
         if (connReq.getSessionId() != 0) {
+        	factory.closeSessionWithoutWakeup(connReq.getSessionId());
             setSessionId(connReq.getSessionId());
             zk.reopenSession(this, sessionId, passwd, sessionTimeout);
             LOG.info("Renewing session 0x" + Long.toHexString(sessionId));

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

@@ -208,7 +208,7 @@ public class PrepRequestProcessor extends Thread implements RequestProcessor {
             case OpCode.create:
                 txnHeader = new TxnHeader(request.sessionId, request.cxid, zks
                         .getNextZxid(), zks.getTime(), OpCode.create);
-                zks.sessionTracker.checkSession(request.sessionId);
+                zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
                 CreateRequest createRequest = new CreateRequest();
                 ZooKeeperServer.byteBuffer2Record(request.request,
                         createRequest);
@@ -270,7 +270,7 @@ public class PrepRequestProcessor extends Thread implements RequestProcessor {
             case OpCode.delete:
                 txnHeader = new TxnHeader(request.sessionId, request.cxid, zks
                         .getNextZxid(), zks.getTime(), OpCode.delete);
-                zks.sessionTracker.checkSession(request.sessionId);
+                zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
                 DeleteRequest deleteRequest = new DeleteRequest();
                 ZooKeeperServer.byteBuffer2Record(request.request,
                         deleteRequest);
@@ -304,7 +304,7 @@ public class PrepRequestProcessor extends Thread implements RequestProcessor {
             case OpCode.setData:
                 txnHeader = new TxnHeader(request.sessionId, request.cxid, zks
                         .getNextZxid(), zks.getTime(), OpCode.setData);
-                zks.sessionTracker.checkSession(request.sessionId);
+                zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
                 SetDataRequest setDataRequest = new SetDataRequest();
                 ZooKeeperServer.byteBuffer2Record(request.request,
                         setDataRequest);
@@ -326,7 +326,7 @@ public class PrepRequestProcessor extends Thread implements RequestProcessor {
             case OpCode.setACL:
                 txnHeader = new TxnHeader(request.sessionId, request.cxid, zks
                         .getNextZxid(), zks.getTime(), OpCode.setACL);
-                zks.sessionTracker.checkSession(request.sessionId);
+                zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
                 SetACLRequest setAclRequest = new SetACLRequest();
                 ZooKeeperServer.byteBuffer2Record(request.request,
                         setAclRequest);
@@ -356,10 +356,12 @@ public class PrepRequestProcessor extends Thread implements RequestProcessor {
                 txn = new CreateSessionTxn(to);
                 request.request.rewind();
                 zks.sessionTracker.addSession(request.sessionId, to);
+                zks.sessionTracker.setOwner(request.sessionId, request.getOwner());
                 break;
             case OpCode.closeSession:
                 txnHeader = new TxnHeader(request.sessionId, request.cxid, zks
                         .getNextZxid(), zks.getTime(), OpCode.closeSession);
+                zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
                 HashSet<String> es = zks.dataTree
                         .getEphemerals(request.sessionId);
                 synchronized (zks.outstandingChanges) {
@@ -386,6 +388,7 @@ public class PrepRequestProcessor extends Thread implements RequestProcessor {
             case OpCode.getChildren:
             case OpCode.ping:
             case OpCode.setWatches:
+            	zks.sessionTracker.checkSession(request.sessionId, request.getOwner());
                 break;
             }
         } catch (KeeperException e) {
@@ -393,6 +396,8 @@ public class PrepRequestProcessor extends Thread implements RequestProcessor {
                 txnHeader.setType(OpCode.error);
                 txn = new ErrorTxn(e.code().intValue());
             }
+            LOG.warn("Got exception when processing " + request.toString(), e);
+            request.setException(e);
         } catch (Exception e) {
             // log at error level as we are returning a marshalling
             // error to the user

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

@@ -23,6 +23,7 @@ import java.util.List;
 
 import org.apache.jute.Record;
 import org.apache.log4j.Logger;
+import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.ZooDefs.OpCode;
 import org.apache.zookeeper.data.Id;
 import org.apache.zookeeper.txn.TxnHeader;
@@ -74,6 +75,18 @@ public class Request {
     public List<Id> authInfo;
 
     public long createTime = System.currentTimeMillis();
+    
+    private Object owner;
+    
+    private KeeperException e;
+    
+    public Object getOwner() {
+        return owner;
+    }
+    
+    public void setOwner(Object owner) {
+        this.owner = owner;
+    }
 
     /**
      * is the packet type a valid packet in zookeeper
@@ -194,4 +207,12 @@ public class Request {
 
         return sb.toString();
     }
+
+    public void setException(KeeperException e) {
+        this.e = e;
+    }
+	
+    public KeeperException getException() {
+        return e;
+    }
 }

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

@@ -19,6 +19,8 @@
 package org.apache.zookeeper.server;
 
 import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.KeeperException.SessionExpiredException;
+import org.apache.zookeeper.KeeperException.SessionMovedException;
 
 /**
  * This is the basic interface that ZooKeeperServer uses to track sessions. The
@@ -54,5 +56,7 @@ public interface SessionTracker {
      */
     void removeSession(long sessionId);
 
-    void checkSession(long sessionId) throws KeeperException.SessionExpiredException;
+    void checkSession(long sessionId, Object owner) throws KeeperException.SessionExpiredException, SessionMovedException;
+
+    void setOwner(long id, Object owner) throws SessionExpiredException;
 }

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

@@ -28,6 +28,8 @@ import java.util.concurrent.ConcurrentHashMap;
 
 import org.apache.log4j.Logger;
 import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.KeeperException.SessionExpiredException;
+import org.apache.zookeeper.KeeperException.SessionMovedException;
 
 /**
  * This is a full featured SessionTracker. It tracks session in grouped by tick
@@ -57,6 +59,8 @@ public class SessionTrackerImpl extends Thread implements SessionTracker {
         long tickTime;
 
         long sessionId;
+        
+        Object owner;
     }
 
     public static long initializeNextSession(long id) {
@@ -216,9 +220,24 @@ public class SessionTrackerImpl extends Thread implements SessionTracker {
         touchSession(id, sessionTimeout);
     }
 
-    public void checkSession(long sessionId) throws KeeperException.SessionExpiredException {
-        if (sessionsById.get(sessionId) == null) {
+    synchronized public void checkSession(long sessionId, Object owner) throws KeeperException.SessionExpiredException, KeeperException.SessionMovedException {
+        Session session = sessionsById.get(sessionId);
+		if (session == null) {
             throw new KeeperException.SessionExpiredException();
         }
+		if (session.owner == null) {
+			session.owner = owner;
+		} else if (session.owner != owner) {
+			throw new KeeperException.SessionMovedException();
+		}
     }
+
+	@Override
+	synchronized public void setOwner(long id, Object owner) throws SessionExpiredException {
+		Session session = sessionsById.get(id);
+		if (session == null) {
+            throw new KeeperException.SessionExpiredException();
+        }
+		session.owner = owner;
+	}
 }

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

@@ -534,7 +534,7 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
      * @param xid
      * @param bb
      */
-    public void submitRequest(ServerCnxn cnxn, long sessionId, int type,
+    private 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);

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

@@ -33,7 +33,9 @@ import org.apache.jute.BinaryInputArchive;
 import org.apache.jute.BinaryOutputArchive;
 import org.apache.jute.Record;
 import org.apache.log4j.Logger;
+import org.apache.zookeeper.KeeperException.SessionExpiredException;
 import org.apache.zookeeper.ZooDefs.OpCode;
+import org.apache.zookeeper.server.Request;
 import org.apache.zookeeper.server.ZooTrace;
 import org.apache.zookeeper.server.quorum.Leader.Proposal;
 import org.apache.zookeeper.server.util.SerializeUtils;
@@ -351,6 +353,13 @@ public class FollowerHandler extends Thread {
                     DataOutputStream dos = new DataOutputStream(bos);
                     dos.writeLong(id);
                     boolean valid = leader.zk.touch(id, to);
+                    if (valid) {
+                        try {
+                            leader.zk.setOwner(id, this);
+                        } catch (SessionExpiredException e) {
+                            LOG.error("Somehow session " + Long.toHexString(id) + " expired right after being renewed! (impossible)", e);
+                        }
+                    }
                     if (LOG.isTraceEnabled()) {
                         ZooTrace.logTraceMessage(LOG,
                                                  ZooTrace.SESSION_TRACE_MASK,
@@ -371,8 +380,9 @@ public class FollowerHandler extends Thread {
                      	leader.zk.submitRequest(new FollowerSyncRequest(this, sessionId, cxid, type, bb,
                                 qp.getAuthinfo()));
                     } else {
-                    leader.zk.submitRequest(null, sessionId, type, cxid, bb,
-                            qp.getAuthinfo());
+                        Request si = new Request(null, sessionId, cxid, type, bb, qp.getAuthinfo());
+                        si.setOwner(this);
+                        leader.zk.submitRequest(si);
                     }
                     break;
                 default:

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

@@ -79,7 +79,11 @@ public class FollowerSessionTracker implements SessionTracker {
         return (nextSessionId++);
     }
 
-    public void checkSession(long sessionId)  {
+    public void checkSession(long sessionId, Object owner)  {
+        // Nothing to do here. Sessions are checked at the Leader
+    }
+    
+    public void setOwner(long sessionId, Object owner) {
         // Nothing to do here. Sessions are checked at the Leader
     }
 }

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

@@ -20,11 +20,13 @@ package org.apache.zookeeper.server.quorum;
 
 import java.io.IOException;
 
+import org.apache.zookeeper.KeeperException.SessionExpiredException;
 import org.apache.zookeeper.jmx.MBeanRegistry;
 import org.apache.zookeeper.server.DataTreeBean;
 import org.apache.zookeeper.server.FinalRequestProcessor;
 import org.apache.zookeeper.server.PrepRequestProcessor;
 import org.apache.zookeeper.server.RequestProcessor;
+import org.apache.zookeeper.server.ServerCnxn;
 import org.apache.zookeeper.server.SessionTrackerImpl;
 import org.apache.zookeeper.server.ZooKeeperServer;
 import org.apache.zookeeper.server.persistence.FileTxnSnapLog;
@@ -151,4 +153,19 @@ public class LeaderZooKeeperServer extends ZooKeeperServer {
     public String getState() {
         return "leader";
     }
+
+    public void setOwner(long id, Object owner) throws SessionExpiredException {
+        sessionTracker.setOwner(id, owner);
+    }
+
+    @Override
+    protected void revalidateSession(ServerCnxn cnxn, long sessionId,
+        int sessionTimeout) throws IOException, InterruptedException {
+        super.revalidateSession(cnxn, sessionId, sessionTimeout);
+        try {
+            setOwner(sessionId, this);
+        } catch (SessionExpiredException e) {
+            // this is ok, it just means that the session revalidation failed.
+        }
+    }
 }

src/java/test/org/apache/zookeeper/server/quorum/QuorumPeerMainTest.java

@@ -201,7 +201,7 @@ public class QuorumPeerMainTest extends TestCase implements Watcher {
                 ClientBase.waitForServerUp("localhost:" + CLIENT_PORT_QP1,
                         5000);
 
-            assertFalse(isup);
+            assertFalse("Server never came up", isup);
             
             q1.shutdown();
 

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

@@ -189,7 +189,7 @@ public class AsyncTest extends TestCase
             verifyThreadTerminated(hammers[i], 60000);
         }
         // before restart
-        QuorumTest qt = new QuorumTest();
+        QuorumBase qt = new QuorumBase();
         qt.verifyRootOfAllServersMatch(qb.hostPort);
         tearDown();
 

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

@@ -18,14 +18,23 @@
 
 package org.apache.zookeeper.test;
 import java.io.IOException;
+import java.util.ArrayList;
+
+import junit.framework.TestCase;
 
 import org.apache.log4j.Logger;
+import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.ZooDefs;
+import org.apache.zookeeper.ZooKeeper;
 import org.junit.Before;
 import org.junit.Test;
 
-public class QuorumTest extends QuorumBase {
+public class QuorumTest extends TestCase {
     private static final Logger LOG = Logger.getLogger(QuorumTest.class);
+    public static final long CONNECTION_TIMEOUT = ClientTest.CONNECTION_TIMEOUT;
     private QuorumBase qb = new QuorumBase();
     private final ClientTest ct = new ClientTest();
     
@@ -86,6 +95,44 @@ public class QuorumTest extends QuorumBase {
     {
         ct.testMutipleWatcherObjs();
     }
+    
+    @Test
+    /**
+     * Connect to two different servers with two different handles using the same session and
+     * make sure we cannot do any changes.
+     */
+    public void testSessionMove() throws IOException, InterruptedException, KeeperException {
+        String hps[] = qb.hostPort.split(",");
+        ZooKeeper zk = new DisconnectableZooKeeper(hps[0], ClientBase.CONNECTION_TIMEOUT, new Watcher() {
+            public void process(WatchedEvent event) {
+        }});
+        zk.create("/t1", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
+        // This should stomp the zk handle
+        ZooKeeper zknew = new DisconnectableZooKeeper(hps[1], ClientBase.CONNECTION_TIMEOUT, new Watcher() {
+            public void process(WatchedEvent event) {
+            }}, zk.getSessionId(), zk.getSessionPasswd());
+        zknew.create("/t2", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
+        try {
+            zk.create("/t3", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
+            fail("Should have lost the connection");
+        } catch(KeeperException.SessionMovedException e) {
+        }
+        
+        ArrayList<ZooKeeper> toClose = new ArrayList<ZooKeeper>();
+        toClose.add(zknew);
+        // Let's just make sure it can still move
+        for(int i = 0; i < 10; i++) {
+            zknew = new DisconnectableZooKeeper(hps[1], ClientBase.CONNECTION_TIMEOUT, new Watcher() {
+                public void process(WatchedEvent event) {
+                }}, zk.getSessionId(), zk.getSessionPasswd());
+            toClose.add(zknew);
+            zknew.create("/t-"+i, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
+        }
+	for(ZooKeeper z: toClose) {
+            z.close();
+        }
+        zk.close();
+    }
 
     // skip superhammer and clientcleanup as they are too expensive for quorum
 }

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

@@ -179,6 +179,31 @@ public class SessionTest extends TestCase implements Watcher {
         zk.close();
     }
 
+    /**
+     * Make sure that we cannot have two connections with the same
+     * session id.
+     * 
+     * @throws IOException
+     * @throws InterruptedException
+     * @throws KeeperException
+     */
+    @Test
+    public void testSessionMove() throws IOException, InterruptedException, KeeperException {
+    	ZooKeeper zk = createClient();
+    	zk.getChildren("/", false);
+    	// This should stomp the zk handle
+    	ZooKeeper zknew = new DisconnectableZooKeeper(HOSTPORT, CONNECTION_TIMEOUT, this,
+                   zk.getSessionId(),
+                   zk.getSessionPasswd());
+    	zknew.getChildren("/", false);
+    	try {
+    	    zk.getChildren("/", false);
+    	    fail("Should have lost the connection");
+    	} catch(KeeperException.ConnectionLossException e) {
+    	}
+    	zknew.close();
+    	zk.close();
+    }
     @Test
     /**
      * This test makes sure that duplicate state changes are not communicated