ZOOKEEPER-484. Clients get SESSION MOVED exception when switching from follower to a leader. (mahadev)

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

CHANGES.txt

@@ -48,6 +48,9 @@ BUGFIXES:
 
   ZOOKEEPER-311. handle small path lengths in zoo_create() (chris barroch via breed)
 
+  ZOOKEEPER-484.  Clients get SESSION MOVED exception when switching from
+  follower to a leader. (mahadev)
+
 IMPROVEMENTS:
   ZOOKEEPER-473. cleanup junit tests to eliminate false positives due to
   "socket reuse" and failure to close client (phunt via mahadev)

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

@@ -70,10 +70,7 @@ 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;
 
@@ -545,7 +542,7 @@ public class NIOServerCnxn implements Watcher, ServerCnxn {
             return;
         } else {
             Request si = new Request(this, sessionId, h.getXid(), h.getType(), incomingBuffer, authInfo);
-            si.setOwner(me);
+            si.setOwner(ServerCnxn.me);
             zk.submitRequest(si);
         }
         if (h.getXid() >= 0) {

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

@@ -356,7 +356,7 @@ 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());
+                zks.setOwner(request.sessionId, request.getOwner());
                 break;
             case OpCode.closeSession:
                 txnHeader = new TxnHeader(request.sessionId, request.cxid, zks

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

@@ -55,6 +55,10 @@ public interface ServerCnxn extends Watcher {
     
     final static ByteBuffer imok = ByteBuffer.wrap("imok".getBytes());
 
+    // This is just an arbitrary object to represent requests issued by
+    // (aka owned by) this class
+    final public static Object me = new Object();
+
     public abstract int getSessionTimeout();
 
     public abstract void close();

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

@@ -38,6 +38,7 @@ import org.apache.jute.OutputArchive;
 import org.apache.jute.Record;
 import org.apache.log4j.Logger;
 import org.apache.zookeeper.Environment;
+import org.apache.zookeeper.KeeperException.SessionExpiredException;
 import org.apache.zookeeper.ZooDefs.OpCode;
 import org.apache.zookeeper.data.ACL;
 import org.apache.zookeeper.data.Id;
@@ -500,6 +501,16 @@ public class ZooKeeperServer implements SessionExpirer, ServerStats.Provider {
         return sessionId;
     }
 
+    /**
+     * set the owner of this session as owner
+     * @param id the session id
+     * @param owner the owner of the session
+     * @throws SessionExpiredException
+     */
+    public void setOwner(long id, Object owner) throws SessionExpiredException {
+        sessionTracker.setOwner(id, owner);
+    }
+
     protected void revalidateSession(ServerCnxn cnxn, long sessionId,
             int sessionTimeout) throws IOException, InterruptedException {
         boolean rc = sessionTracker.touchSession(sessionId, sessionTimeout);

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

@@ -355,6 +355,9 @@ public class FollowerHandler extends Thread {
                     boolean valid = leader.zk.touch(id, to);
                     if (valid) {
                         try {
+                            //set the session owner
+                            // as the follower that
+                            // owns the session
                             leader.zk.setOwner(id, this);
                         } catch (SessionExpiredException e) {
                             LOG.error("Somehow session " + Long.toHexString(id) + " expired right after being renewed! (impossible)", e);

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

@@ -154,16 +154,14 @@ public class LeaderZooKeeperServer extends ZooKeeperServer {
         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);
+            // setowner as the leader itself, unless updated
+            // via the follower handlers
+            setOwner(sessionId, ServerCnxn.me);
         } catch (SessionExpiredException e) {
             // this is ok, it just means that the session revalidation failed.
         }

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

@@ -17,6 +17,8 @@
  */
 
 package org.apache.zookeeper.test;
+import static org.apache.zookeeper.test.ClientBase.CONNECTION_TIMEOUT;
+
 import java.io.IOException;
 import java.util.ArrayList;
 
@@ -27,6 +29,7 @@ import org.apache.zookeeper.WatchedEvent;
 import org.apache.zookeeper.Watcher;
 import org.apache.zookeeper.ZooDefs;
 import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.ZooDefs.Ids;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -97,6 +100,40 @@ public class QuorumTest extends QuorumBase {
         ct.testMutipleWatcherObjs();
     }
 
+    /**
+     * Make sure that we can change sessions 
+     *  from follower to leader.
+     *
+     * @throws IOException
+     * @throws InterruptedException
+     * @throws KeeperException
+     */
+    @Test
+    public void testSessionMoved() throws IOException, InterruptedException, KeeperException {
+        String hostPorts[] = qb.hostPort.split(",");
+        ZooKeeper zk = new DisconnectableZooKeeper(hostPorts[0], ClientBase.CONNECTION_TIMEOUT, new Watcher() {
+            public void process(WatchedEvent event) {
+            }});
+        zk.create("/sessionMoveTest", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
+        // we want to loop through the list twice
+        for(int i = 0; i < hostPorts.length*2; i++) {
+            // This should stomp the zk handle
+            ZooKeeper zknew = new DisconnectableZooKeeper(hostPorts[(i+1)%hostPorts.length], ClientBase.CONNECTION_TIMEOUT, 
+                    new Watcher() {public void process(WatchedEvent event) {
+                    }},
+                    zk.getSessionId(),
+                    zk.getSessionPasswd());
+            zknew.setData("/", new byte[1], -1);
+            try {
+                zk.setData("/", new byte[1], -1);
+                fail("Should have lost the connection");
+            } catch(KeeperException.SessionMovedException e) {
+            }
+            //zk.close();
+            zk = zknew;
+        }
+        zk.close();
+    }
     @Test
     /**
      * Connect to two different servers with two different handles using the same session and

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

@@ -224,19 +224,24 @@ public class SessionTest extends TestCase implements Watcher {
      */
     @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) {
+        String hostPorts[] = HOSTPORT.split(",");
+        ZooKeeper zk = new DisconnectableZooKeeper(hostPorts[0], CONNECTION_TIMEOUT, this);
+        zk.create("/sessionMoveTest", new byte[0], Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
+        // we want to loop through the list twice
+        for(int i = 0; i < hostPorts.length*2; i++) {
+            // This should stomp the zk handle
+            ZooKeeper zknew = new DisconnectableZooKeeper(hostPorts[(i+1)%hostPorts.length], CONNECTION_TIMEOUT, this,
+                    zk.getSessionId(),
+                    zk.getSessionPasswd());
+            zknew.setData("/", new byte[1], -1);
+            try {
+                zk.setData("/", new byte[1], -1);
+                fail("Should have lost the connection");
+            } catch(KeeperException.ConnectionLossException e) {
+            }
+            //zk.close();
+            zk = zknew;
         }
-        zknew.close();
         zk.close();
     }
     @Test