ZOOKEEPER-2786: Flaky test: org.apache.zookeeper.test.ClientTest.testNonExistingOpCode

As hanm noticed and mentioned on the JIRA this flaky test can still fail. The reason is that we can reconnect before calling `watcher.waitForDisconnected(5000);` We need to make sure that we only process a single event from the watcher. Using a `synchronized (watcher)` provides that guarantee here.

This issue can be reproduced by dropping a Thread.sleep before `watcher.waitForDisconnected(5000);`.

Author: Abraham Fine <afine@apache.org>

Reviewers: Michael Han <hanm@apache.org>

Closes #286 from afine/ZOOKEEPER-2786_3.5_fix

(cherry picked from commit 74e45a0d635702f1a1f57adffdaba9acdc9efee2)
Signed-off-by: Michael Han <hanm@apache.org>

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

@@ -18,12 +18,11 @@
 
 package org.apache.zookeeper.test;
 
-import static org.junit.Assert.fail;
-
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 
@@ -33,6 +32,7 @@ import org.apache.zookeeper.KeeperException.Code;
 import org.apache.zookeeper.KeeperException.InvalidACLException;
 import org.apache.zookeeper.TestableZooKeeper;
 import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
 import org.apache.zookeeper.Watcher.Event.EventType;
 import org.apache.zookeeper.Watcher.Event.KeeperState;
 import org.apache.zookeeper.ZooDefs.Ids;
@@ -828,8 +828,16 @@ public class ClientTest extends ClientBase {
      */
     @Test
     public void testNonExistingOpCode() throws Exception  {
-        CountdownWatcher watcher = new CountdownWatcher();
-        TestableZooKeeper zk = createClient(watcher);
+        final CountDownLatch clientDisconnected = new CountDownLatch(1);
+        Watcher watcher = new Watcher() {
+            @Override
+            public synchronized void process(WatchedEvent event) {
+                if (event.getState() == KeeperState.Disconnected) {
+                    clientDisconnected.countDown();
+                }
+            }
+        };
+        TestableZooKeeper zk = new TestableZooKeeper(hostPort, CONNECTION_TIMEOUT, watcher);
 
         final String path = "/m1";
 
@@ -839,12 +847,14 @@ public class ClientTest extends ClientBase {
         request.setPath(path);
         request.setWatch(false);
         ExistsResponse response = new ExistsResponse();
+
         ReplyHeader r = zk.submitRequest(h, request, response, null);
 
         Assert.assertEquals(r.getErr(), Code.UNIMPLEMENTED.intValue());
 
         // Sending a nonexisting opcode should cause the server to disconnect
-        watcher.waitForDisconnected(5000);
+        Assert.assertTrue("failed to disconnect",
+                clientDisconnected.await(5000, TimeUnit.MILLISECONDS));
     }
 
     @Test