ZOOKEEPER-1046: Creating a new sequential node results in a ZNODEEXISTS error

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

CHANGES.txt

@@ -239,6 +239,7 @@ BUGFIXES:
   ZOOKEEPER-1103. In QuorumTest, use the same "for ( .. try { break }
   catch { } )" pattern in testFollowersStartAfterLeaders as in
   testSessionMove. (Eugene Koontz via phunt)
+  ZOOKEEPER-1046. Creating a new sequential node results in a ZNODEEXISTS error. (breed via camille) 
 
 IMPROVEMENTS:
   ZOOKEEPER-724. Improve junit test integration - log harness information 

src/contrib/loggraph/src/java/org/apache/zookeeper/graph/TxnLogSource.java

@@ -180,9 +180,8 @@ public class TxnLogSource implements LogSource {
 		    throw new IOException("CRC doesn't match " + crcValue +
 					  " vs " + crc.getValue());
 		}
-		InputArchive iab = BinaryInputArchive.getArchive(new ByteArrayInputStream(bytes));
 		TxnHeader hdr = new TxnHeader();
-		Record r = SerializeUtils.deserializeTxn(iab, hdr);
+		Record r = SerializeUtils.deserializeTxn(bytes, hdr);
 
 		switch (hdr.getType()) {
 		case OpCode.createSession: {
@@ -328,9 +327,8 @@ public class TxnLogSource implements LogSource {
 		if (logStream.readByte("EOR") != 'B') {
 		    throw new EOFException("Last transaction was partial.");
 		}
-		InputArchive iab = BinaryInputArchive.getArchive(new ByteArrayInputStream(bytes));
 		TxnHeader hdr = new TxnHeader();
-		Record r = SerializeUtils.deserializeTxn(iab, hdr);
+		Record r = SerializeUtils.deserializeTxn(bytes, hdr);
 		
 		if (starttime == 0) {
 		    starttime = hdr.getTime();

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

@@ -134,7 +134,6 @@ public class DataNode implements Record {
     synchronized public void copyStat(Stat to) {
         to.setAversion(stat.getAversion());
         to.setCtime(stat.getCtime());
-        to.setCversion(stat.getCversion());
         to.setCzxid(stat.getCzxid());
         to.setMtime(stat.getMtime());
         to.setMzxid(stat.getMzxid());
@@ -142,11 +141,15 @@ public class DataNode implements Record {
         to.setVersion(stat.getVersion());
         to.setEphemeralOwner(stat.getEphemeralOwner());
         to.setDataLength(data == null ? 0 : data.length);
-        if (this.children == null) {
-            to.setNumChildren(0);
-        } else {
-            to.setNumChildren(children.size());
+        int numChildren = 0;
+        if (this.children != null) {
+            numChildren = children.size();
         }
+        // when we do the Cversion we need to translate from the count of the creates
+        // to the count of the changes (v3 semantics)
+        // for every create there is a delete except for the children still present
+        to.setCversion(stat.getCversion()*2 - numChildren);
+        to.setNumChildren(numChildren);
     }
 
     synchronized public void deserialize(InputArchive archive, String tag)

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

@@ -446,7 +446,7 @@ public class DataTree {
      * @throws KeeperException
      */
     public String createNode(String path, byte data[], List<ACL> acl,
-            long ephemeralOwner, long zxid, long time)
+            long ephemeralOwner, int parentCVersion, long zxid, long time)
             throws KeeperException.NoNodeException,
             KeeperException.NodeExistsException {
         int lastSlash = path.lastIndexOf('/');
@@ -472,9 +472,12 @@ public class DataTree {
                     throw new KeeperException.NodeExistsException();
                 }
             }
-            int cver = parent.stat.getCversion();
-            cver++;
-            parent.stat.setCversion(cver);
+            
+            if (parentCVersion == -1) {
+                parentCVersion = parent.stat.getCversion();
+                parentCVersion++;
+            }    
+            parent.stat.setCversion(parentCVersion);
             parent.stat.setPzxid(zxid);
             Long longval = convertAcls(acl);
             DataNode child = new DataNode(parent, data, longval, stat);
@@ -542,7 +545,6 @@ public class DataTree {
         }
         synchronized (parent) {
             parent.removeChild(childName);
-            parent.stat.setCversion(parent.stat.getCversion() + 1);
             parent.stat.setPzxid(zxid);
             long eowner = node.stat.getEphemeralOwner();
             if (eowner != 0) {
@@ -770,6 +772,7 @@ public class DataTree {
                             createTxn.getData(),
                             createTxn.getAcl(),
                             createTxn.getEphemeral() ? header.getClientId() : 0,
+                            createTxn.getParentCVersion(),
                             header.getZxid(), header.getTime());
                     break;
                 case OpCode.delete:

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

@@ -85,10 +85,8 @@ public class LogFormatter {
                 throw new IOException("CRC doesn't match " + crcValue +
                         " vs " + crc.getValue());
             }
-            InputArchive iab = BinaryInputArchive
-                                .getArchive(new ByteArrayInputStream(bytes));
             TxnHeader hdr = new TxnHeader();
-            SerializeUtils.deserializeTxn(iab, hdr);
+            Record txn = SerializeUtils.deserializeTxn(bytes, hdr);
             System.out.println(DateFormat.getDateTimeInstance(DateFormat.SHORT,
                     DateFormat.LONG).format(new Date(hdr.getTime()))
                     + " session 0x"
@@ -97,7 +95,7 @@ public class LogFormatter {
                     + Long.toHexString(hdr.getCxid())
                     + " zxid 0x"
                     + Long.toHexString(hdr.getZxid())
-                    + " " + TraceFormatter.op2String(hdr.getType()));
+                    + " " + TraceFormatter.op2String(hdr.getType()) + " " + txn);
             if (logStream.readByte("EOR") != 'B') {
                 LOG.error("Last transaction was partial.");
                 throw new EOFException("Last transaction was partial.");

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

@@ -256,17 +256,17 @@ public class PrepRequestProcessor extends Thread implements RequestProcessor {
                 if (ephemeralParent) {
                     throw new KeeperException.NoChildrenForEphemeralsException(path);
                 }
+                int newCversion = parentRecord.stat.getCversion()+1;
                 txn = new CreateTxn(path, createRequest.getData(),
                         createRequest.getAcl(),
-                        createMode.isEphemeral());
+                        createMode.isEphemeral(), newCversion);
                 StatPersisted s = new StatPersisted();
                 if (createMode.isEphemeral()) {
                     s.setEphemeralOwner(request.sessionId);
                 }
                 parentRecord = parentRecord.duplicate(txnHeader.getZxid());
                 parentRecord.childCount++;
-                parentRecord.stat
-                        .setCversion(parentRecord.stat.getCversion() + 1);
+                parentRecord.stat.setCversion(newCversion);
                 addChangeRecord(parentRecord);
                 addChangeRecord(new ChangeRecord(txnHeader.getZxid(), path, s,
                         0, createRequest.getAcl()));
@@ -300,8 +300,6 @@ public class PrepRequestProcessor extends Thread implements RequestProcessor {
                 txn = new DeleteTxn(path);
                 parentRecord = parentRecord.duplicate(txnHeader.getZxid());
                 parentRecord.childCount--;
-                parentRecord.stat
-                        .setCversion(parentRecord.stat.getCversion() + 1);
                 addChangeRecord(parentRecord);
                 addChangeRecord(new ChangeRecord(txnHeader.getZxid(), path,
                         null, -1, null));

src/java/main/org/apache/zookeeper/server/persistence/FileTxnLog.java

@@ -569,10 +569,8 @@ public class FileTxnLog implements TxnLog {
                     throw new IOException(CRC_ERROR);
                 if (bytes == null || bytes.length == 0)
                     return false;
-                InputArchive iab = BinaryInputArchive
-                                    .getArchive(new ByteArrayInputStream(bytes));
                 hdr = new TxnHeader();
-                record = SerializeUtils.deserializeTxn(iab, hdr);
+                record = SerializeUtils.deserializeTxn(bytes, hdr);
             } catch (EOFException e) {
                 LOG.debug("EOF excepton " + e);
                 inputStream.close();

src/java/main/org/apache/zookeeper/server/persistence/FileTxnSnapLog.java

@@ -33,6 +33,7 @@ import org.apache.zookeeper.server.Request;
 import org.apache.zookeeper.server.ZooTrace;
 import org.apache.zookeeper.server.persistence.TxnLog.TxnIterator;
 import org.apache.zookeeper.txn.CreateSessionTxn;
+import org.apache.zookeeper.txn.CreateTxn;
 import org.apache.zookeeper.txn.TxnHeader;
 import org.apache.zookeeper.KeeperException.Code;
 import org.apache.zookeeper.KeeperException.NoNodeException;
@@ -212,10 +213,9 @@ public class FileTxnSnapLog {
          * Note, such failures on DT should be seen only during
          * restore.
          */
-        if ((hdr.getType() == OpCode.delete &&
-                 rc.err == Code.NONODE.intValue()) ||
-            (hdr.getType() == OpCode.create &&
-                rc.err == Code.NODEEXISTS.intValue())) {
+        if ((hdr.getType() == OpCode.create &&
+                rc.err == Code.NODEEXISTS.intValue()) &&
+                ((CreateTxn)txn).getParentCVersion() == -1) {
             LOG.debug("Failed Txn: " + hdr.getType() + " path:" +
                   rc.path + " err: " + rc.err);
             int lastSlash = rc.path.lastIndexOf('/');

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

@@ -113,9 +113,7 @@ public class Follower extends Learner{
             break;
         case Leader.PROPOSAL:            
             TxnHeader hdr = new TxnHeader();
-            BinaryInputArchive ia = BinaryInputArchive
-            .getArchive(new ByteArrayInputStream(qp.getData()));
-            Record txn = SerializeUtils.deserializeTxn(ia, hdr);
+            Record txn = SerializeUtils.deserializeTxn(qp.getData(), hdr);
             if (hdr.getZxid() != lastQueued + 1) {
                 LOG.warn("Got zxid 0x"
                         + Long.toHexString(hdr.getZxid())

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

@@ -361,9 +361,7 @@ public class Learner {
                 case Leader.PROPOSAL:
                     PacketInFlight pif = new PacketInFlight();
                     pif.hdr = new TxnHeader();
-                    BinaryInputArchive ia = BinaryInputArchive
-                            .getArchive(new ByteArrayInputStream(qp.getData()));
-                    pif.rec     = SerializeUtils.deserializeTxn(ia, pif.hdr);
+                    pif.rec = SerializeUtils.deserializeTxn(qp.getData(), pif.hdr);
                     if (pif.hdr.    getZxid() != lastQueued + 1) {
                     LOG.warn("Got zxid 0x"
                             + Long.toHexString(pif.hdr.getZxid())
@@ -384,9 +382,7 @@ public class Learner {
                     break;
                 case Leader.INFORM:
                     TxnHeader hdr = new TxnHeader();
-                    ia = BinaryInputArchive
-                            .getArchive(new ByteArrayInputStream(qp.getData()));
-                    Record txn = SerializeUtils.deserializeTxn(ia, hdr);
+                    Record txn = SerializeUtils.deserializeTxn(qp.getData(), hdr);
                     zk.getZKDatabase().processTxn(hdr, txn);
                     break;
                 case Leader.UPTODATE:

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

@@ -190,11 +190,9 @@ public class LearnerHandler extends Thread {
             break;
         case Leader.PROPOSAL:
             type = "PROPOSAL";
-            BinaryInputArchive ia = BinaryInputArchive
-                    .getArchive(new ByteArrayInputStream(p.getData()));
             TxnHeader hdr = new TxnHeader();
             try {
-                txn = SerializeUtils.deserializeTxn(ia, hdr);
+                txn = SerializeUtils.deserializeTxn(p.getData(), hdr);
                 // mess = "transaction: " + txn.toString();
             } catch (IOException e) {
                 LOG.warn("Unexpected exception",e);

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

@@ -118,9 +118,7 @@ public class Observer extends Learner{
             break;
         case Leader.INFORM:            
             TxnHeader hdr = new TxnHeader();
-            BinaryInputArchive ia = BinaryInputArchive
-                    .getArchive(new ByteArrayInputStream(qp.getData()));
-            Record txn = SerializeUtils.deserializeTxn(ia, hdr);
+            Record txn = SerializeUtils.deserializeTxn(qp.getData(), hdr);
             Request request = new Request (null, hdr.getClientId(), 
                                            hdr.getCxid(),
                                            hdr.getType(), null, null);

src/java/main/org/apache/zookeeper/server/upgrade/UpgradeSnapShotV1.java

@@ -115,10 +115,8 @@ public class UpgradeSnapShotV1 implements UpgradeSnapShot {
                     // empty transaction
                     throw new EOFException();
                 }
-                InputArchive ia = BinaryInputArchive
-                        .getArchive(new ByteArrayInputStream(bytes));
                 TxnHeader hdr = new TxnHeader();
-                Record txn = SerializeUtils.deserializeTxn(ia, hdr);
+                Record txn = SerializeUtils.deserializeTxn(bytes, hdr);
                 if (logStream.readByte("EOR") != 'B') {
                     LOG.warn("Last transaction was partial.");
                     throw new EOFException("Last transaction was partial.");

src/java/main/org/apache/zookeeper/server/util/SerializeUtils.java

@@ -18,6 +18,8 @@
 
 package org.apache.zookeeper.server.util;
 
+import java.io.ByteArrayInputStream;
+import java.io.EOFException;
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.Map;
@@ -26,6 +28,7 @@ import java.util.Map.Entry;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.jute.BinaryInputArchive;
 import org.apache.jute.InputArchive;
 import org.apache.jute.OutputArchive;
 import org.apache.jute.Record;
@@ -34,6 +37,7 @@ import org.apache.zookeeper.server.DataTree;
 import org.apache.zookeeper.server.ZooTrace;
 import org.apache.zookeeper.txn.CreateSessionTxn;
 import org.apache.zookeeper.txn.CreateTxn;
+import org.apache.zookeeper.txn.CreateTxnV0;
 import org.apache.zookeeper.txn.DeleteTxn;
 import org.apache.zookeeper.txn.ErrorTxn;
 import org.apache.zookeeper.txn.SetACLTxn;
@@ -43,9 +47,13 @@ import org.apache.zookeeper.txn.TxnHeader;
 public class SerializeUtils {
     private static final Logger LOG = LoggerFactory.getLogger(SerializeUtils.class);
     
-    public static Record deserializeTxn(InputArchive ia, TxnHeader hdr)
+    public static Record deserializeTxn(byte txnBytes[], TxnHeader hdr)
             throws IOException {
+        final ByteArrayInputStream bais = new ByteArrayInputStream(txnBytes);
+        InputArchive ia = BinaryInputArchive.getArchive(bais);
+
         hdr.deserialize(ia, "hdr");
+        bais.mark(bais.available());
         Record txn = null;
         switch (hdr.getType()) {
         case OpCode.createSession:
@@ -72,7 +80,26 @@ public class SerializeUtils {
             break;
         }
         if (txn != null) {
-            txn.deserialize(ia, "txn");
+            try {
+                txn.deserialize(ia, "txn");
+            } catch(EOFException e) {
+                // perhaps this is a V0 Create
+                if (hdr.getType() == OpCode.create) {
+                    CreateTxn create = (CreateTxn)txn;
+                    bais.reset();
+                    CreateTxnV0 createv0 = new CreateTxnV0();
+                    createv0.deserialize(ia, "txn");
+                    // cool now make it V1. a -1 parentCVersion will
+                    // trigger fixup processing in processTxn
+                    create.setPath(createv0.getPath());
+                    create.setData(createv0.getData());
+                    create.setAcl(createv0.getAcl());
+                    create.setEphemeral(createv0.getEphemeral());
+                    create.setParentCVersion(-1);
+                } else {
+                    throw e;
+                }
+            }
         }
         return txn;
     }

src/java/test/org/apache/zookeeper/server/DataTreeUnitTest.java

@@ -53,7 +53,7 @@ public class DataTreeUnitTest extends ZKTestCase {
         // set a watch on the root node
         dt.getChildren("/", new Stat(), watcher);
         // add a new node, should trigger a watch
-        dt.createNode("/xyz", new byte[0], null, 0, 1, 1);
+        dt.createNode("/xyz", new byte[0], null, 0, dt.getNode("/").stat.getCversion()+1, 1, 1);
         Assert.assertFalse("Root node watch not triggered",!watcher.fired);
     }
 

src/java/test/org/apache/zookeeper/server/DeserializationPerfTest.java

@@ -40,7 +40,7 @@ public class DeserializationPerfTest extends ZKTestCase {
         int count;
         {
             DataTree tree = new DataTree();
-            SerializationPerfTest.createNodes(tree, "/", depth, width, new byte[len]);
+            SerializationPerfTest.createNodes(tree, "/", depth, tree.getNode("/").stat.getCversion(), width, new byte[len]);
             count = tree.getNodeCount();
 
             ByteArrayOutputStream baos = new ByteArrayOutputStream();

src/java/test/org/apache/zookeeper/server/SerializationPerfTest.java

@@ -38,9 +38,9 @@ public class SerializationPerfTest extends ZKTestCase {
     }
 
     static int createNodes(DataTree tree, String path, int depth,
-            int childcount, byte[] data) throws KeeperException.NodeExistsException, KeeperException.NoNodeException {
+            int childcount, int parentCVersion, byte[] data) throws KeeperException.NodeExistsException, KeeperException.NoNodeException {
         path += "node" + depth;
-        tree.createNode(path, data, null, -1, 1, 1);
+        tree.createNode(path, data, null, -1, ++parentCVersion, 1, 1);
 
         if (--depth == 0) {
             return 1;
@@ -50,7 +50,7 @@ public class SerializationPerfTest extends ZKTestCase {
 
         int count = 1;
         for (int i = 0; i < childcount; i++) {
-            count += createNodes(tree, path + i, depth, childcount, data);
+            count += createNodes(tree, path + i, depth, childcount, 1, data);
         }
 
         return count;
@@ -59,7 +59,7 @@ public class SerializationPerfTest extends ZKTestCase {
     private static void serializeTree(int depth, int width, int len)
             throws InterruptedException, IOException, KeeperException.NodeExistsException, KeeperException.NoNodeException {
         DataTree tree = new DataTree();
-        createNodes(tree, "/", depth, width, new byte[len]);
+        createNodes(tree, "/", depth, width, tree.getNode("/").stat.getCversion(), new byte[len]);
         int count = tree.getNodeCount();
 
         BinaryOutputArchive oa =

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

@@ -59,7 +59,7 @@ public class DataTreeTest extends ZKTestCase {
         // set a watch on the root node
         dt.getChildren("/", new Stat(), watcher);
         // add a new node, should trigger a watch
-        dt.createNode("/xyz", new byte[0], null, 0, 1, 1);
+        dt.createNode("/xyz", new byte[0], null, 0, dt.getNode("/").stat.getCversion()+1, 1, 1);
         Assert.assertFalse("Root node watch not triggered",!watcher.fired);
     }
 
@@ -68,7 +68,7 @@ public class DataTreeTest extends ZKTestCase {
      */
     @Test
     public void testIncrementCversion() throws Exception {
-        dt.createNode("/test", new byte[0], null, 0, 1, 1);
+        dt.createNode("/test", new byte[0], null, 0, dt.getNode("/").stat.getCversion()+1, 1, 1);
         DataNode zk = dt.getNode("/test");
         long prevCversion = zk.stat.getCversion();
         long prevPzxid = zk.stat.getPzxid();

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

@@ -127,9 +127,9 @@ public class LoadFromLogTest extends ZKTestCase implements  Watcher {
         File tmpDir = ClientBase.createTmpDir();
         FileTxnSnapLog logFile = new FileTxnSnapLog(tmpDir, tmpDir);
         DataTree dt = new DataTree();
-        dt.createNode("/test", new byte[0], null, 0, 1, 1);
+        dt.createNode("/test", new byte[0], null, 0, -1, 1, 1);
         for (count = 1; count <= 3; count++) {
-            dt.createNode("/test/" + count, new byte[0], null, 0, count,
+            dt.createNode("/test/" + count, new byte[0], null, 0, -1, count,
                     System.currentTimeMillis());
         }
         DataNode zk = dt.getNode("/test");
@@ -139,8 +139,9 @@ public class LoadFromLogTest extends ZKTestCase implements  Watcher {
         doOp(logFile, OpCode.create, "/test/" + (count - 1), dt, zk);
 
         // Make delete fo fail, then verify cversion.
-        LOG.info("Attempting to delete " + "/test/" + (count + 1));
-        doOp(logFile, OpCode.delete, "/test/" + (count + 1), dt, zk);
+        // this doesn't happen anymore, we only set the cversion on create
+        // LOG.info("Attempting to delete " + "/test/" + (count + 1));
+        // doOp(logFile, OpCode.delete, "/test/" + (count + 1), dt, zk);
     }
     /*
      * Does create/delete depending on the type and verifies
@@ -151,7 +152,7 @@ public class LoadFromLogTest extends ZKTestCase implements  Watcher {
         int lastSlash = path.lastIndexOf('/');
         String parentName = path.substring(0, lastSlash);
 
-        long prevCversion = parent.stat.getCversion();
+        int prevCversion = parent.stat.getCversion();
         long prevPzxid = parent.stat.getPzxid();
         List<String> child = dt.getChildren(parentName, null, null);
         String childStr = "";
@@ -160,7 +161,7 @@ public class LoadFromLogTest extends ZKTestCase implements  Watcher {
         }
         LOG.info("Children: " + childStr + " for " + parentName);
         LOG.info("(cverions, pzxid): " + prevCversion + ", " + prevPzxid);
-
+        
         Record txn = null;
         TxnHeader txnHeader = null;
         if (type == OpCode.delete) {
@@ -170,7 +171,7 @@ public class LoadFromLogTest extends ZKTestCase implements  Watcher {
         } else if (type == OpCode.create) {
             txnHeader = new TxnHeader(0xabcd, 0x123, prevPzxid + 1,
                     System.currentTimeMillis(), OpCode.create);
-            txn = new CreateTxn(path, new byte[0], null, false);
+            txn = new CreateTxn(path, new byte[0], null, false, -1);
         }
         logFile.processTransaction(txnHeader, dt, null, txn);
 
@@ -183,7 +184,7 @@ public class LoadFromLogTest extends ZKTestCase implements  Watcher {
         }
         LOG.info("Children: " + childStr + " for " + parentName);
         LOG.info("(cverions, pzxid): " +newCversion + ", " + newPzxid);
-        Assert.assertTrue("<cversion, pzxid> verification failed. Expected: <" +
+        Assert.assertTrue(type + " <cversion, pzxid> verification failed. Expected: <" +
                 (prevCversion + 1) + ", " + (prevPzxid + 1) + ">, found: <" +
                 newCversion + ", " + newPzxid + ">",
                 (newCversion == prevCversion + 1 && newPzxid == prevPzxid + 1));
@@ -198,7 +199,7 @@ public class LoadFromLogTest extends ZKTestCase implements  Watcher {
         FileTxnLog txnLog = new FileTxnLog(tmpDir);
         TxnHeader txnHeader = new TxnHeader(0xabcd, 0x123, 0x123,
               System.currentTimeMillis(), OpCode.create);
-        Record txn = new CreateTxn("/Test", new byte[0], null, false);
+        Record txn = new CreateTxn("/Test", new byte[0], null, false, 1);
         txnLog.append(txnHeader, txn);
         FileInputStream in = new FileInputStream(tmpDir.getPath() + "/log." +
               Long.toHexString(txnHeader.getZxid()));

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

@@ -32,12 +32,13 @@ import org.apache.log4j.Level;
 import org.apache.log4j.Logger;
 import org.apache.log4j.WriterAppender;
 import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.KeeperException.NotReadOnlyException;
 import org.apache.zookeeper.WatchedEvent;
 import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.Watcher.Event.KeeperState;
 import org.apache.zookeeper.ZooDefs;
 import org.apache.zookeeper.ZooKeeper;
-import org.apache.zookeeper.KeeperException.NotReadOnlyException;
-import org.apache.zookeeper.Watcher.Event.KeeperState;
 import org.apache.zookeeper.ZooKeeper.States;
 import org.junit.After;
 import org.junit.Before;
@@ -113,10 +114,18 @@ public class ReadOnlyModeTest extends QuorumBase {
                         states.add(event.getState());
                     }
                 }, true);
-
-        Thread.sleep(1000);
-        zk.create("/test", "test".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,
-                CreateMode.PERSISTENT);
+        boolean success = false;
+        for (int i = 0; i < 30; i++) {
+            try {
+                zk.create("/test", "test".getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE,
+                        CreateMode.PERSISTENT);
+                success=true;
+                break;
+            } catch(KeeperException.ConnectionLossException e) {
+                Thread.sleep(1000);               
+            }            
+        }
+        Assert.assertTrue("Did not succeed in connecting in 30s", success);
 
         // kill peer and wait no more than 5 seconds for read-only server
         // to be started (which should take one tickTime (2 seconds))