diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/DataTree.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/DataTree.java
index b99bf632da6..6dbe94332c1 100644
--- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/DataTree.java
+++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/DataTree.java
@@ -1962,4 +1962,18 @@ public static StatPersisted createStat(long zxid, long time, long ephemeralOwner
         stat.setEphemeralOwner(ephemeralOwner);
         return stat;
     }
+
+    // for test only
+    static StatPersisted createStat(int version) {
+        StatPersisted stat = new StatPersisted();
+        stat.setCtime(0);
+        stat.setMtime(0);
+        stat.setCzxid(0);
+        stat.setMzxid(0);
+        stat.setPzxid(0);
+        stat.setVersion(version);
+        stat.setAversion(0);
+        stat.setEphemeralOwner(0);
+        return stat;
+    }
 }
diff --git a/zookeeper-server/src/main/java/org/apache/zookeeper/server/PrepRequestProcessor.java b/zookeeper-server/src/main/java/org/apache/zookeeper/server/PrepRequestProcessor.java
index 28d1cc97b55..274862ee31f 100644
--- a/zookeeper-server/src/main/java/org/apache/zookeeper/server/PrepRequestProcessor.java
+++ b/zookeeper-server/src/main/java/org/apache/zookeeper/server/PrepRequestProcessor.java
@@ -738,7 +738,18 @@ private static int checkAndIncVersion(int currentVersion, int expectedVersion, S
         if (expectedVersion != -1 && expectedVersion != currentVersion) {
             throw new KeeperException.BadVersionException(path);
         }
-        return currentVersion + 1;
+        // Increase once more when going back to -1 from Integer.MIN_VALUE. Otherwise, the client will
+        // receive a new data version -1. And Now, if the client wants to check the data version, it can
+        // only pass -1 as the next expected version, but -1 as the expected version means do not check
+        // the data version. So the client is unable to express the expected manner.
+        //
+        // See also https://issues.apache.org/jira/browse/ZOOKEEPER-4743.
+        int nextVersion = currentVersion + 1;
+        if (nextVersion == -1) {
+            return 0;
+        } else {
+            return nextVersion;
+        }
     }
 
     /**
diff --git a/zookeeper-server/src/test/java/org/apache/zookeeper/server/PrepRequestProcessorTest.java b/zookeeper-server/src/test/java/org/apache/zookeeper/server/PrepRequestProcessorTest.java
index b10258fb5b3..e2416415807 100644
--- a/zookeeper-server/src/test/java/org/apache/zookeeper/server/PrepRequestProcessorTest.java
+++ b/zookeeper-server/src/test/java/org/apache/zookeeper/server/PrepRequestProcessorTest.java
@@ -45,6 +45,7 @@
 import org.apache.zookeeper.ZooDefs.Ids;
 import org.apache.zookeeper.ZooDefs.OpCode;
 import org.apache.zookeeper.data.Id;
+import org.apache.zookeeper.data.Stat;
 import org.apache.zookeeper.proto.CreateRequest;
 import org.apache.zookeeper.proto.ReconfigRequest;
 import org.apache.zookeeper.proto.RequestHeader;
@@ -59,6 +60,7 @@
 import org.apache.zookeeper.server.quorum.flexible.QuorumVerifier;
 import org.apache.zookeeper.test.ClientBase;
 import org.apache.zookeeper.txn.ErrorTxn;
+import org.apache.zookeeper.txn.SetDataTxn;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -391,4 +393,110 @@ public Set<Long> localSessions() {
             return Collections.emptySet();
         }
     }
+
+    @Test
+    public void testCheckAndIncVersion() throws Exception {
+        zks.getZKDatabase().dataTree.createNode("/foo", new byte[0], Ids.OPEN_ACL_UNSAFE, 0, 0, 0, 0);
+        pLatch = new CountDownLatch(1);
+        processor = new PrepRequestProcessor(zks, new MyRequestProcessor());
+
+        SetDataRequest record = new SetDataRequest("/foo", new byte[0], 0);
+        Request req = createRequest(record, OpCode.setData, false);
+        processor.pRequest(req);
+        pLatch.await();
+        assertEquals(OpCode.setData, outcome.getHdr().getType());
+        assertTrue(outcome.getTxn() instanceof SetDataTxn);
+        SetDataTxn setDataTxn = (SetDataTxn) outcome.getTxn();
+        assertEquals(1, setDataTxn.getVersion());
+    }
+
+    @Test
+    public void testCheckAndIncVersionOverflow() throws Exception {
+        Stat customStat = new Stat();
+        customStat.setVersion(Integer.MAX_VALUE);
+        zks.getZKDatabase().dataTree.createNode("/foo", new byte[0], Ids.OPEN_ACL_UNSAFE, 0, 0, 0, 0);
+        DataNode node = zks.getZKDatabase().dataTree.getNode("/foo");
+        node.stat = DataTree.createStat(Integer.MAX_VALUE);
+
+        pLatch = new CountDownLatch(1);
+        processor = new PrepRequestProcessor(zks, new MyRequestProcessor());
+
+        SetDataRequest record = new SetDataRequest("/foo", new byte[0], Integer.MAX_VALUE);
+        Request req = createRequest(record, OpCode.setData, false);
+        processor.pRequest(req);
+        pLatch.await();
+        assertEquals(OpCode.setData, outcome.getHdr().getType());
+        assertTrue(outcome.getTxn() instanceof SetDataTxn);
+        SetDataTxn setDataTxn = (SetDataTxn) outcome.getTxn();
+        assertEquals(Integer.MIN_VALUE, setDataTxn.getVersion());
+    }
+    @Test
+    public void testCheckAndIncVersionWithNegativeNumber() throws Exception {
+        zks.getZKDatabase().dataTree.createNode("/foo", new byte[0], Ids.OPEN_ACL_UNSAFE, 0, 0, 0, 0);
+        DataNode node = zks.getZKDatabase().dataTree.getNode("/foo");
+        node.stat = DataTree.createStat(Integer.MIN_VALUE);
+
+        pLatch = new CountDownLatch(1);
+        processor = new PrepRequestProcessor(zks, new MyRequestProcessor());
+
+        SetDataRequest record = new SetDataRequest("/foo", new byte[0], Integer.MIN_VALUE);
+        Request req = createRequest(record, OpCode.setData, false);
+        processor.pRequest(req);
+        pLatch.await();
+        assertEquals(OpCode.setData, outcome.getHdr().getType());
+        assertTrue(outcome.getTxn() instanceof SetDataTxn);
+        SetDataTxn setDataTxn = (SetDataTxn) outcome.getTxn();
+        assertEquals(Integer.MIN_VALUE + 1, setDataTxn.getVersion());
+    }
+
+    @Test
+    public void testCheckAndIncToZeroFromNegativeTwo() throws Exception {
+        zks.getZKDatabase().dataTree.createNode("/foo", new byte[0], Ids.OPEN_ACL_UNSAFE, 0, 0, 0, 0);
+        DataNode node = zks.getZKDatabase().dataTree.getNode("/foo");
+        node.stat = DataTree.createStat(-2);
+
+        pLatch = new CountDownLatch(1);
+        processor = new PrepRequestProcessor(zks, new MyRequestProcessor());
+
+        SetDataRequest record = new SetDataRequest("/foo", new byte[0], -2);
+        Request req = createRequest(record, OpCode.setData, false);
+        processor.pRequest(req);
+        pLatch.await();
+        assertEquals(OpCode.setData, outcome.getHdr().getType());
+        assertTrue(outcome.getTxn() instanceof SetDataTxn);
+        SetDataTxn setDataTxn = (SetDataTxn) outcome.getTxn();
+        assertEquals(0, setDataTxn.getVersion());
+    }
+
+    @Test
+    public void testCheckAndIncSkipEqualityCheck() throws Exception {
+        zks.getZKDatabase().dataTree.createNode("/foo", new byte[0], Ids.OPEN_ACL_UNSAFE, 0, 0, 0, 0);
+        DataNode node = zks.getZKDatabase().dataTree.getNode("/foo");
+
+        pLatch = new CountDownLatch(1);
+        processor = new PrepRequestProcessor(zks, new MyRequestProcessor());
+
+        SetDataRequest record = new SetDataRequest("/foo", new byte[0], -1);
+        Request req = createRequest(record, OpCode.setData, false);
+        processor.pRequest(req);
+        pLatch.await();
+        assertEquals(OpCode.setData, outcome.getHdr().getType());
+        assertTrue(outcome.getTxn() instanceof SetDataTxn);
+        SetDataTxn setDataTxn = (SetDataTxn) outcome.getTxn();
+        assertEquals(1, setDataTxn.getVersion());
+    }
+
+    @Test
+    public void testCheckAndIncWithBadVersion() throws Exception {
+        zks.getZKDatabase().dataTree.createNode("/foo", new byte[0], Ids.OPEN_ACL_UNSAFE, 0, 0, 0, 0);
+        pLatch = new CountDownLatch(1);
+        processor = new PrepRequestProcessor(zks, new MyRequestProcessor());
+
+        SetDataRequest record = new SetDataRequest("/foo", new byte[0], 1);
+        Request req = createRequest(record, OpCode.setData, false);
+        processor.pRequest(req);
+        pLatch.await();
+        assertEquals(OpCode.error, outcome.getHdr().getType());
+        assertEquals(KeeperException.Code.BADVERSION, outcome.getException().code());
+    }
 }