ZOOKEEPER-961. Watch recovery after disconnection when connection string contains a prefix. (Matthias Spycher via mahadev)

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

CHANGES.txt

@@ -319,6 +319,9 @@ BUGFIXES:
 
   ZOOKEEPER-1125. Intermittent java core test failures. (Vishar Kher via mahadev)
 
+  ZOOKEEPER-961. Watch recovery after disconnection when connection string contains a prefix.
+  (Matthias Spycher via mahadev)
+
 IMPROVEMENTS:
   ZOOKEEPER-724. Improve junit test integration - log harness information 
   (phunt via mahadev)

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

@@ -18,27 +18,29 @@
 
 package org.apache.zookeeper;
 
+import java.io.BufferedReader;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.io.InputStreamReader;
 import java.lang.Thread.UncaughtExceptionHandler;
+import java.net.ConnectException;
 import java.net.InetSocketAddress;
+import java.net.Socket;
 import java.net.SocketAddress;
 import java.nio.ByteBuffer;
 import java.util.LinkedList;
+import java.util.List;
 import java.util.Random;
 import java.util.Set;
 import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.LinkedBlockingQueue;
-import java.io.BufferedReader;
-import java.io.InputStreamReader;
-import java.net.ConnectException;
-import java.net.Socket;
+
+import javax.security.auth.login.LoginException;
+import javax.security.sasl.SaslException;
 
 import org.apache.jute.BinaryInputArchive;
 import org.apache.jute.BinaryOutputArchive;
 import org.apache.jute.Record;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.apache.zookeeper.AsyncCallback.ACLCallback;
 import org.apache.zookeeper.AsyncCallback.Children2Callback;
 import org.apache.zookeeper.AsyncCallback.ChildrenCallback;
@@ -71,9 +73,8 @@ import org.apache.zookeeper.proto.SetWatches;
 import org.apache.zookeeper.proto.WatcherEvent;
 import org.apache.zookeeper.server.ByteBufferInputStream;
 import org.apache.zookeeper.server.ZooTrace;
-
-import javax.security.auth.login.LoginException;
-import javax.security.sasl.SaslException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * This class manages the socket i/o for the client. ClientCnxn maintains a list
@@ -748,8 +749,13 @@ public class ClientCnxn {
                     String serverPath = event.getPath();
                     if(serverPath.compareTo(chrootPath)==0)
                         event.setPath("/");
-                    else
+                    else if (serverPath.length() > chrootPath.length())
                         event.setPath(serverPath.substring(chrootPath.length()));
+                    else {
+                    	LOG.warn("Got server path " + event.getPath()
+                    			+ " which is too short for chroot path "
+                    			+ chrootPath);
+                    }
                 }
 
                 WatchedEvent we = new WatchedEvent(event);
@@ -843,19 +849,22 @@ public class ClientCnxn {
                 // Only send if there's a pending watch
                 // TODO: here we have the only remaining use of zooKeeper in
                 // this class. It's to be eliminated!
-                if (!disableAutoWatchReset
-                        && (!zooKeeper.getDataWatches().isEmpty()
-                                || !zooKeeper.getExistWatches().isEmpty() || !zooKeeper
-                                .getChildWatches().isEmpty())) {
-                    SetWatches sw = new SetWatches(lastZxid,
-                            zooKeeper.getDataWatches(),
-                            zooKeeper.getExistWatches(),
-                            zooKeeper.getChildWatches());
-                    RequestHeader h = new RequestHeader();
-                    h.setType(ZooDefs.OpCode.setWatches);
-                    h.setXid(-8);
-                    Packet packet = new Packet(h, new ReplyHeader(), sw, null, null);
-                    outgoingQueue.addFirst(packet);
+                if (!disableAutoWatchReset) {
+                    List<String> dataWatches = zooKeeper.getDataWatches();
+                    List<String> existWatches = zooKeeper.getExistWatches();
+                    List<String> childWatches = zooKeeper.getChildWatches();
+                    if (!dataWatches.isEmpty()
+                                || !existWatches.isEmpty() || !childWatches.isEmpty()) {
+                        SetWatches sw = new SetWatches(lastZxid,
+                                prependChroot(dataWatches),
+                                prependChroot(existWatches),
+                                prependChroot(childWatches));
+                        RequestHeader h = new RequestHeader();
+                        h.setType(ZooDefs.OpCode.setWatches);
+                        h.setXid(-8);
+                        Packet packet = new Packet(h, new ReplyHeader(), sw, null, null);
+                        outgoingQueue.addFirst(packet);
+                    }
                 }
 
                 for (AuthData id : authInfo) {
@@ -873,6 +882,23 @@ public class ClientCnxn {
             }
         }
 
+        private List<String> prependChroot(List<String> paths) {
+            if (chrootPath != null && !paths.isEmpty()) {
+                for (int i = 0; i < paths.size(); ++i) {
+                    String clientPath = paths.get(i);
+                    String serverPath;
+                    // handle clientPath = "/"
+                    if (clientPath.length() == 1) {
+                        serverPath = chrootPath;
+                    } else {
+                        serverPath = chrootPath + clientPath;
+                    }
+                    paths.set(i, serverPath);
+                }
+            }
+            return paths;
+        }
+
         private void sendPing() {
             lastPingSentNs = System.nanoTime();
             RequestHeader h = new RequestHeader(-2, OpCode.ping);
@@ -888,13 +914,6 @@ public class ClientCnxn {
         private int pingRwTimeout = minPingRwTimeout;
 
         private void startConnect() throws IOException {
-            if(!isFirstConnect){
-                try {
-                    Thread.sleep(r.nextInt(1000));
-                } catch (InterruptedException e) {
-                    LOG.warn("Unexpected exception", e);
-                }
-            }
             state = States.CONNECTING;
 
             InetSocketAddress addr;
@@ -937,8 +956,15 @@ public class ClientCnxn {
             while (state.isAlive()) {
                 try {
                     if (!clientCnxnSocket.isConnected()) {
+                        if(!isFirstConnect){
+                            try {
+                                Thread.sleep(r.nextInt(1000));
+                            } catch (InterruptedException e) {
+                                LOG.warn("Unexpected exception", e);
+                            }
+                        }
                         // don't re-establish connection if we are closing
-                        if (closing) {
+                        if (closing || !state.isAlive()) {
                             break;
                         }
                         startConnect();

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

@@ -0,0 +1,176 @@
+/**
+ * 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.test;
+
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.Watcher.Event.EventType;
+import org.apache.zookeeper.ZooDefs.Ids;
+import org.junit.Assert;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class DisconnectedWatcherTest extends ClientBase {
+    protected static final Logger LOG = LoggerFactory.getLogger(DisconnectedWatcherTest.class);
+    final int TIMEOUT = 5000;
+
+    private class MyWatcher extends CountdownWatcher {
+        LinkedBlockingQueue<WatchedEvent> events =
+            new LinkedBlockingQueue<WatchedEvent>();
+
+        public void process(WatchedEvent event) {
+            super.process(event);
+            if (event.getType() != Event.EventType.None) {
+                try {
+                    events.put(event);
+                } catch (InterruptedException e) {
+                    LOG.warn("ignoring interrupt during event.put");
+                }
+            }
+        }
+    }
+
+    // @see jira issue ZOOKEEPER-961
+    
+    @Test
+    public void testChildWatcherAutoResetWithChroot() throws Exception {
+        ZooKeeper zk1 = createClient();
+
+        zk1.create("/ch1", null, Ids.OPEN_ACL_UNSAFE,
+                    CreateMode.PERSISTENT);
+
+        MyWatcher watcher = new MyWatcher();
+        ZooKeeper zk2 = createClient(watcher, hostPort + "/ch1");
+        zk2.getChildren("/", true );
+
+        // this call shouldn't trigger any error or watch
+        zk1.create("/youdontmatter1", null, Ids.OPEN_ACL_UNSAFE,
+                CreateMode.PERSISTENT);
+
+        // this should trigger the watch
+        zk1.create("/ch1/youshouldmatter1", null, Ids.OPEN_ACL_UNSAFE,
+                CreateMode.PERSISTENT);
+        WatchedEvent e = watcher.events.poll(TIMEOUT, TimeUnit.MILLISECONDS);
+        Assert.assertNotNull(e);
+        Assert.assertEquals(EventType.NodeChildrenChanged, e.getType());
+        Assert.assertEquals("/", e.getPath());
+
+        MyWatcher childWatcher = new MyWatcher();
+        zk2.getChildren("/", childWatcher);
+        
+        stopServer();
+        watcher.waitForDisconnected(3000);
+        startServer();
+        watcher.waitForConnected(3000);
+
+        // this should trigger the watch
+        zk1.create("/ch1/youshouldmatter2", null, Ids.OPEN_ACL_UNSAFE,
+                CreateMode.PERSISTENT);
+        e = childWatcher.events.poll(TIMEOUT, TimeUnit.MILLISECONDS);
+        Assert.assertNotNull(e);
+        Assert.assertEquals(EventType.NodeChildrenChanged, e.getType());
+        Assert.assertEquals("/", e.getPath());
+    }
+    
+    @Test
+    public void testDefaultWatcherAutoResetWithChroot() throws Exception {
+        ZooKeeper zk1 = createClient();
+
+        zk1.create("/ch1", null, Ids.OPEN_ACL_UNSAFE,
+                    CreateMode.PERSISTENT);
+
+        MyWatcher watcher = new MyWatcher();
+        ZooKeeper zk2 = createClient(watcher, hostPort + "/ch1");
+        zk2.getChildren("/", true );
+
+        // this call shouldn't trigger any error or watch
+        zk1.create("/youdontmatter1", null, Ids.OPEN_ACL_UNSAFE,
+                CreateMode.PERSISTENT);
+
+        // this should trigger the watch
+        zk1.create("/ch1/youshouldmatter1", null, Ids.OPEN_ACL_UNSAFE,
+                CreateMode.PERSISTENT);
+        WatchedEvent e = watcher.events.poll(TIMEOUT, TimeUnit.MILLISECONDS);
+        Assert.assertNotNull(e);
+        Assert.assertEquals(EventType.NodeChildrenChanged, e.getType());
+        Assert.assertEquals("/", e.getPath());
+
+        zk2.getChildren("/", true );
+
+        stopServer();
+        watcher.waitForDisconnected(3000);
+        startServer();
+        watcher.waitForConnected(3000);
+
+        // this should trigger the watch
+        zk1.create("/ch1/youshouldmatter2", null, Ids.OPEN_ACL_UNSAFE,
+                CreateMode.PERSISTENT);
+        e = watcher.events.poll(TIMEOUT, TimeUnit.MILLISECONDS);
+        Assert.assertNotNull(e);
+        Assert.assertEquals(EventType.NodeChildrenChanged, e.getType());
+        Assert.assertEquals("/", e.getPath());
+    }
+    
+    @Test
+    public void testDeepChildWatcherAutoResetWithChroot() throws Exception {
+        ZooKeeper zk1 = createClient();
+
+        zk1.create("/ch1", null, Ids.OPEN_ACL_UNSAFE,
+                CreateMode.PERSISTENT);
+        zk1.create("/ch1/here", null, Ids.OPEN_ACL_UNSAFE,
+                CreateMode.PERSISTENT);
+        zk1.create("/ch1/here/we", null, Ids.OPEN_ACL_UNSAFE,
+                CreateMode.PERSISTENT);
+        zk1.create("/ch1/here/we/are", null, Ids.OPEN_ACL_UNSAFE,
+                CreateMode.PERSISTENT);
+
+        MyWatcher watcher = new MyWatcher();
+        ZooKeeper zk2 = createClient(watcher, hostPort + "/ch1/here/we");
+        zk2.getChildren("/are", true );
+
+        // this should trigger the watch
+        zk1.create("/ch1/here/we/are/now", null, Ids.OPEN_ACL_UNSAFE,
+                CreateMode.PERSISTENT);
+        WatchedEvent e = watcher.events.poll(TIMEOUT, TimeUnit.MILLISECONDS);
+        Assert.assertNotNull(e);
+        Assert.assertEquals(EventType.NodeChildrenChanged, e.getType());
+        Assert.assertEquals("/are", e.getPath());
+
+        MyWatcher childWatcher = new MyWatcher();
+        zk2.getChildren("/are", childWatcher);
+        
+        stopServer();
+        watcher.waitForDisconnected(3000);
+        startServer();
+        watcher.waitForConnected(3000);
+
+        // this should trigger the watch
+        zk1.create("/ch1/here/we/are/again", null, Ids.OPEN_ACL_UNSAFE,
+                CreateMode.PERSISTENT);
+        e = childWatcher.events.poll(TIMEOUT, TimeUnit.MILLISECONDS);
+        Assert.assertNotNull(e);
+        Assert.assertEquals(EventType.NodeChildrenChanged, e.getType());
+        Assert.assertEquals("/are", e.getPath());
+    }
+}