ZOOKEEPER-5092: sequence numbers don't increment past Integer.MAX_VALUE

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

@@ -467,12 +467,11 @@ public void createNode(final String path, byte[] data, List<ACL> acl, long ephem
                 parentCVersion = parent.stat.getCversion();
                 parentCVersion++;
             }
-            // There is possibility that we'll replay txns for a node which
-            // was created and then deleted in the fuzzy range, and it's not
-            // exist in the snapshot, so replay the creation might revert the
-            // cversion and pzxid, need to check and only update when it's
-            // larger.
-            if (parentCVersion > parent.stat.getCversion()) {
+            // Fuzzy snapshot replay can apply an older create after a newer one.
+            // Only move cversion forward. The signed counter wrapping from
+            // Integer.MAX_VALUE to Integer.MIN_VALUE is a forward step
+            // (ZOOKEEPER-5092), not a replay of an older value.
+            if (isCversionAdvance(parent.stat.getCversion(), parentCVersion)) {
                 parent.stat.setCversion(parentCVersion);
                 parent.stat.setPzxid(zxid);
             }
@@ -1563,7 +1562,7 @@ public void setCversionPzxid(String path, int newCversion, long zxid) throws NoN
             if (newCversion == -1) {
                 newCversion = node.stat.getCversion() + 1;
             }
-            if (newCversion > node.stat.getCversion()) {
+            if (isCversionAdvance(node.stat.getCversion(), newCversion)) {
                 nodes.preChange(path, node);
                 node.stat.setCversion(newCversion);
                 node.stat.setPzxid(zxid);
@@ -1572,6 +1571,16 @@ public void setCversionPzxid(String path, int newCversion, long zxid) throws NoN
         }
     }
 
+    /**
+     * True when {@code proposed} is a later cversion than {@code current}.
+     * The signed 32-bit wrap from {@link Integer#MAX_VALUE} to
+     * {@link Integer#MIN_VALUE} is an advance. Any other smaller value is not:
+     * fuzzy snapshot replay must not move the counter backwards.
+     */
+    static boolean isCversionAdvance(int current, int proposed) {
+        return proposed > current || (current == Integer.MAX_VALUE && proposed == Integer.MIN_VALUE);
+    }
+
     public boolean containsWatcher(String path, WatcherType type, Watcher watcher) {
         boolean containsWatcher = false;
         switch (type) {

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

@@ -156,6 +156,34 @@ public void testIncrementCversion() throws Exception {
         }
     }
 
+    /**
+     * ZOOKEEPER-5092: the next cversion after {@link Integer#MAX_VALUE} is the
+     * signed wrap {@link Integer#MIN_VALUE}. Fuzzy replay must not treat that
+     * wrap as a backwards update.
+     */
+    @Test
+    public void testCversionOverflowAdvances() throws Exception {
+        DataTree dt = new DataTree();
+        dt.createNode("/test", new byte[0], null, 0, 1, 1, 1);
+        dt.setCversionPzxid("/test", Integer.MAX_VALUE, 2);
+        assertEquals(Integer.MAX_VALUE, dt.getNode("/test").stat.getCversion());
+        dt.setCversionPzxid("/test", 1, 9);
+        assertEquals(Integer.MAX_VALUE, dt.getNode("/test").stat.getCversion(),
+            "a smaller cversion must not rewind the counter");
+
+        dt.createNode("/test/child_2147483647", new byte[0], null, 0, Integer.MIN_VALUE, 3, 3);
+
+        assertEquals(Integer.MIN_VALUE, dt.getNode("/test").stat.getCversion());
+        assertEquals(3L, dt.getNode("/test").stat.getPzxid());
+
+        DataTree replay = new DataTree();
+        replay.createNode("/parent", new byte[0], null, 0, 1, 1, 1);
+        replay.setCversionPzxid("/parent", Integer.MAX_VALUE, 2);
+        replay.setCversionPzxid("/parent", Integer.MIN_VALUE, 4);
+        assertEquals(Integer.MIN_VALUE, replay.getNode("/parent").stat.getCversion());
+        assertEquals(4L, replay.getNode("/parent").stat.getPzxid());
+    }
+
     @Test
     public void testNoCversionRevert() throws Exception {
         DataTree dt = new DataTree();