ZOOKEEPER-3125: Only patching the pzxid when it's larger than the current pzxid

This previous fix in #605 has a corner case which might revert the pzxid, it's being fixed when port to 3.5 in #647, update on master as well.

Author: Fangmin Lyu <fangmin@apache.org>

Reviewers: eolivelli@gmail.com

Closes #701 from lvfangmin/ZOOKEEPER-3125-Update

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

@@ -569,7 +569,12 @@ public class DataTree {
         }
         synchronized (parent) {
             parent.removeChild(childName);
-            parent.stat.setPzxid(zxid);
+            // Only update pzxid when the zxid is larger than the current pzxid,
+            // otherwise we might override some higher pzxid set by a create
+            // Txn, which could cause the cversion and pzxid inconsistent
+            if (zxid > parent.stat.getPzxid()) {
+                parent.stat.setPzxid(zxid);
+            }
         }
 
         DataNode node = nodes.get(path);

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

@@ -153,6 +153,31 @@ public class DataTreeTest extends ZKTestCase {
                 (newCversion == prevCversion + 1 && newPzxid == prevPzxid + 1));
     }
 
+    @Test
+    public void testPzxidUpdatedWhenDeletingNonExistNode() throws Exception {
+        DataNode root = dt.getNode("/");
+        long currentPzxid = root.stat.getPzxid();
+
+        // pzxid updated with deleteNode on higher zxid
+        long zxid = currentPzxid + 1;
+        try {
+            dt.deleteNode("/testPzxidUpdatedWhenDeletingNonExistNode", zxid);
+        } catch (NoNodeException e) { /* expected */ }
+        root = dt.getNode("/");
+        currentPzxid = root.stat.getPzxid();
+        Assert.assertEquals(currentPzxid, zxid);
+
+        // pzxid not updated with smaller zxid
+        long prevPzxid = currentPzxid;
+        zxid = prevPzxid - 1;
+        try {
+            dt.deleteNode("/testPzxidUpdatedWhenDeletingNonExistNode", zxid);
+        } catch (NoNodeException e) { /* expected */ }
+        root = dt.getNode("/");
+        currentPzxid = root.stat.getPzxid();
+        Assert.assertEquals(currentPzxid, prevPzxid);
+    }
+
     @Test(timeout = 60000)
     public void testPathTrieClearOnDeserialize() throws Exception {