diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/ClientTest2_2.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/ClientTest2_2.java
new file mode 100644
index 0000000..f794aea
--- /dev/null
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/ClientTest2_2.java
@@ -0,0 +1,35 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.tests.integration.backwardscompatibility;
+
+
+import org.apache.pulsar.tests.integration.topologies.ClientTestBase;
+import org.testng.annotations.Test;
+
+public class ClientTest2_2 extends PulsarStandaloneTestSuite2_2 {
+
+    private final ClientTestBase clientTestBase = new ClientTestBase();
+
+    @Test(dataProvider = "StandaloneServiceUrlAndHttpUrl")
+    public void testResetCursorCompatibility(String serviceUrl, String httpServiceUrl) throws Exception {
+        String topicName = generateTopicName("test-reset-cursor-compatibility", true);
+        clientTestBase.resetCursorCompatibility(serviceUrl, httpServiceUrl, topicName);
+    }
+
+}
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/ClientTest2_3.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/ClientTest2_3.java
new file mode 100644
index 0000000..1bc30fd
--- /dev/null
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/ClientTest2_3.java
@@ -0,0 +1,35 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.tests.integration.backwardscompatibility;
+
+
+import org.apache.pulsar.tests.integration.topologies.ClientTestBase;
+import org.testng.annotations.Test;
+
+public class ClientTest2_3 extends PulsarStandaloneTestSuite2_3 {
+
+    private final ClientTestBase clientTestBase = new ClientTestBase();
+
+    @Test(dataProvider = "StandaloneServiceUrlAndHttpUrl")
+    public void testResetCursorCompatibility(String serviceUrl, String httpServiceUrl) throws Exception {
+        String topicName = generateTopicName("test-reset-cursor-compatibility", true);
+        clientTestBase.resetCursorCompatibility(serviceUrl, httpServiceUrl, topicName);
+    }
+
+}
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/ClientTest2_4.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/ClientTest2_4.java
new file mode 100644
index 0000000..b1d3021
--- /dev/null
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/ClientTest2_4.java
@@ -0,0 +1,35 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.tests.integration.backwardscompatibility;
+
+
+import org.apache.pulsar.tests.integration.topologies.ClientTestBase;
+import org.testng.annotations.Test;
+
+public class ClientTest2_4 extends PulsarStandaloneTestSuite2_4 {
+
+    private final ClientTestBase clientTestBase = new ClientTestBase();
+
+    @Test(dataProvider = "StandaloneServiceUrlAndHttpUrl")
+    public void testResetCursorCompatibility(String serviceUrl, String httpServiceUrl) throws Exception {
+        String topicName = generateTopicName("test-reset-cursor-compatibility", true);
+        clientTestBase.resetCursorCompatibility(serviceUrl, httpServiceUrl, topicName);
+    }
+
+}
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/ClientTest2_5.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/ClientTest2_5.java
new file mode 100644
index 0000000..48a089b
--- /dev/null
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/backwardscompatibility/ClientTest2_5.java
@@ -0,0 +1,35 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.tests.integration.backwardscompatibility;
+
+
+import org.apache.pulsar.tests.integration.topologies.ClientTestBase;
+import org.testng.annotations.Test;
+
+public class ClientTest2_5 extends PulsarStandaloneTestSuite2_5 {
+
+    private final ClientTestBase clientTestBase = new ClientTestBase();
+
+    @Test(dataProvider = "StandaloneServiceUrlAndHttpUrl")
+    public void testResetCursorCompatibility(String serviceUrl, String httpServiceUrl) throws Exception {
+        String topicName = generateTopicName("test-reset-cursor-compatibility", true);
+        clientTestBase.resetCursorCompatibility(serviceUrl, httpServiceUrl, topicName);
+    }
+
+}
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/messaging/PersistentTopicMessagingTest.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/messaging/PersistentTopicMessagingTest.java
index 39b37ac..4bfcde1 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/messaging/PersistentTopicMessagingTest.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/messaging/PersistentTopicMessagingTest.java
@@ -63,4 +63,5 @@
     public void testPartitionedTopicMessagingWithKeyShared(String serviceUrl) throws Exception {
         partitionedTopicSendAndReceiveWithKeyShared(serviceUrl, true);
     }
+
 }
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/messaging/TopicMessagingBase.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/messaging/TopicMessagingBase.java
index 3b37993..c15adca 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/messaging/TopicMessagingBase.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/messaging/TopicMessagingBase.java
@@ -25,7 +25,9 @@
 
 import lombok.Cleanup;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.MessageRoutingMode;
 import org.apache.pulsar.client.api.Producer;
@@ -38,6 +40,7 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.UUID;
+import java.util.concurrent.TimeUnit;
 
 @Slf4j
 public class TopicMessagingBase extends MessagingBase {
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/ClientTestBase.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/ClientTestBase.java
new file mode 100644
index 0000000..3511c48
--- /dev/null
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/ClientTestBase.java
@@ -0,0 +1,74 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.tests.integration.topologies;
+
+import lombok.Cleanup;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.Message;
+import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.Schema;
+
+import java.util.concurrent.TimeUnit;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+
+public class ClientTestBase {
+
+    public void resetCursorCompatibility(String serviceUrl, String serviceHttpUrl, String topicName) throws Exception {
+        final String subName = "my-sub";
+        @Cleanup
+        final PulsarClient pulsarClient = PulsarClient.builder()
+                .serviceUrl(serviceUrl)
+                .build();
+        @Cleanup
+        final PulsarAdmin admin = PulsarAdmin.builder()
+                .serviceHttpUrl(serviceHttpUrl)
+                .build();
+        @Cleanup
+        Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+                .enableBatching(false).topic(topicName).create();
+        @Cleanup
+        Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
+                .topic(topicName).subscriptionName(subName).subscribe();
+        for (int i = 0; i < 50; i++) {
+            producer.send("msg" + i);
+        }
+        Message<String> lastMsg = null;
+        for (int i = 0; i < 10; i++) {
+            lastMsg = consumer.receive();
+            assertNotNull(lastMsg);
+            consumer.acknowledge(lastMsg);
+        }
+
+        admin.topics().resetCursor(topicName, subName, lastMsg.getMessageId());
+        @Cleanup
+        Consumer<String> consumer2 = pulsarClient.newConsumer(Schema.STRING).topic(topicName).subscriptionName(subName).subscribe();
+        Message<String> message = consumer2.receive(1, TimeUnit.SECONDS);
+        assertEquals(message.getMessageId(), lastMsg.getMessageId());
+
+        admin.topics().resetCursorAsync(topicName, subName, lastMsg.getMessageId()).get(3, TimeUnit.SECONDS);
+        @Cleanup
+        Consumer<String> consumer3 = pulsarClient.newConsumer(Schema.STRING).topic(topicName).subscriptionName(subName).subscribe();
+        message = consumer3.receive(1, TimeUnit.SECONDS);
+        assertEquals(message.getMessageId(), lastMsg.getMessageId());
+    }
+}
diff --git a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarStandaloneTestBase.java b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarStandaloneTestBase.java
index 6c3d982..03132f5 100644
--- a/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarStandaloneTestBase.java
+++ b/tests/integration/src/test/java/org/apache/pulsar/tests/integration/topologies/PulsarStandaloneTestBase.java
@@ -53,6 +53,16 @@
         };
     }
 
+    @DataProvider(name = "StandaloneServiceUrlAndHttpUrl")
+    public static Object[][] serviceUrlAndHttpUrl() {
+        return new Object[][] {
+                {
+                        container.getPlainTextServiceUrl(),
+                        container.getHttpServiceUrl(),
+                }
+        };
+    }
+
     protected static Network network;
     protected static StandaloneContainer container;
 
diff --git a/tests/integration/src/test/resources/pulsar-backwards-compatibility.xml b/tests/integration/src/test/resources/pulsar-backwards-compatibility.xml
index b566ca2..f0fc615 100644
--- a/tests/integration/src/test/resources/pulsar-backwards-compatibility.xml
+++ b/tests/integration/src/test/resources/pulsar-backwards-compatibility.xml
@@ -25,6 +25,11 @@
             <class name="org.apache.pulsar.tests.integration.backwardscompatibility.SmokeTest2_2" />
             <class name="org.apache.pulsar.tests.integration.backwardscompatibility.SmokeTest2_3" />
             <class name="org.apache.pulsar.tests.integration.backwardscompatibility.SmokeTest2_4" />
+            <class name="org.apache.pulsar.tests.integration.backwardscompatibility.SmokeTest2_5" />
+            <class name="org.apache.pulsar.tests.integration.backwardscompatibility.ClientTest2_2" />
+            <class name="org.apache.pulsar.tests.integration.backwardscompatibility.ClientTest2_3" />
+            <class name="org.apache.pulsar.tests.integration.backwardscompatibility.ClientTest2_4" />
+            <class name="org.apache.pulsar.tests.integration.backwardscompatibility.ClientTest2_5" />
         </classes>
     </test>
 </suite>
\ No newline at end of file