Admin-api to reset cursor by position (#785)

* Admin-api to reset cursor by position

* Fix: api notes for global topic

* reset cursor using messageId

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/PersistentTopics.java

@@ -19,6 +19,7 @@
 package org.apache.pulsar.broker.admin;
 
 import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.pulsar.broker.cache.ConfigurationCacheService.POLICIES;
 import static org.apache.pulsar.common.util.Codec.decode;
 
 import java.io.IOException;
@@ -79,6 +80,7 @@ import org.apache.pulsar.client.admin.PulsarAdminException.NotFoundException;
 import org.apache.pulsar.client.admin.PulsarAdminException.PreconditionFailedException;
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.impl.MessageIdImpl;
 import org.apache.pulsar.client.util.FutureUtil;
 import org.apache.pulsar.common.api.Commands;
 import org.apache.pulsar.common.api.proto.PulsarApi.KeyValue;
@@ -87,8 +89,6 @@ import org.apache.pulsar.common.compression.CompressionCodec;
 import org.apache.pulsar.common.compression.CompressionCodecProvider;
 import org.apache.pulsar.common.naming.DestinationDomain;
 import org.apache.pulsar.common.naming.DestinationName;
-import org.apache.pulsar.common.naming.NamespaceBundle;
-import org.apache.pulsar.common.naming.NamespaceName;
 import org.apache.pulsar.common.partition.PartitionedTopicMetadata;
 import org.apache.pulsar.common.policies.data.AuthAction;
 import org.apache.pulsar.common.policies.data.AuthPolicies;
@@ -113,7 +113,6 @@ import io.swagger.annotations.Api;
 import io.swagger.annotations.ApiOperation;
 import io.swagger.annotations.ApiResponse;
 import io.swagger.annotations.ApiResponses;
-import static org.apache.pulsar.broker.cache.ConfigurationCacheService.POLICIES;
 
 /**
  */
@@ -906,11 +905,9 @@ public class PersistentTopics extends AdminResource {
 
     @POST
     @Path("/{property}/{cluster}/{namespace}/{destination}/subscription/{subName}/resetcursor/{timestamp}")
-    @ApiOperation(value = "Reset subscription to message position closest to absolute timestamp (in ms).", notes = "There should not be any active consumers on the subscription.")
+    @ApiOperation(value = "Reset subscription to message position closest to absolute timestamp (in ms).", notes = "It fence cursor and disconnects all active consumers before reseting cursor.")
     @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
-            @ApiResponse(code = 404, message = "Topic does not exist"),
-            @ApiResponse(code = 405, message = "Not supported for global and non-persistent topics"),
-            @ApiResponse(code = 412, message = "Subscription has active consumers") })
+            @ApiResponse(code = 404, message = "Topic/Subscription does not exist") })
     public void resetCursor(@PathParam("property") String property, @PathParam("cluster") String cluster,
             @PathParam("namespace") String namespace, @PathParam("destination") @Encoded String destination,
             @PathParam("subName") String subName, @PathParam("timestamp") long timestamp,
@@ -944,8 +941,8 @@ public class PersistentTopics extends AdminResource {
                 log.warn("[{}] [{}] Failed to reset cursor on subscription {} to time {}", clientAppId(), dn, subName,
                         timestamp, partitionException);
                 throw new RestException(Status.PRECONDITION_FAILED, partitionException.getMessage());
-            } else if (numPartException > 0 && log.isDebugEnabled()) {
-                log.debug("[{}][{}] partial errors for reset cursor on subscription {} to time {} - ", clientAppId(),
+            } else if (numPartException > 0) {
+                log.warn("[{}][{}] partial errors for reset cursor on subscription {} to time {} - ", clientAppId(),
                         destination, subName, timestamp, partitionException);
             }
 
@@ -954,6 +951,9 @@ public class PersistentTopics extends AdminResource {
             log.info("[{}][{}] received reset cursor on subscription {} to time {}", clientAppId(), destination,
                     subName, timestamp);
             PersistentTopic topic = (PersistentTopic) getTopicReference(dn);
+            if (topic == null) {
+                throw new RestException(Status.NOT_FOUND, "Topic not found");
+            }
             try {
                 PersistentSubscription sub = topic.getSubscription(subName);
                 checkNotNull(sub);
@@ -967,8 +967,6 @@ public class PersistentTopics extends AdminResource {
                     throw new RestException(Status.NOT_FOUND, "Subscription not found");
                 } else if (e instanceof NotAllowedException) {
                     throw new RestException(Status.METHOD_NOT_ALLOWED, e.getMessage());
-                } else if (t instanceof SubscriptionBusyException) {
-                    throw new RestException(Status.PRECONDITION_FAILED, "Subscription has active connected consumers");
                 } else if (t instanceof SubscriptionInvalidCursorPosition) {
                     throw new RestException(Status.PRECONDITION_FAILED,
                             "Unable to find position for timestamp specified -" + t.getMessage());
@@ -979,6 +977,56 @@ public class PersistentTopics extends AdminResource {
         }
     }
 
+    @POST
+    @Path("/{property}/{cluster}/{namespace}/{destination}/subscription/{subName}/resetcursor")
+    @ApiOperation(value = "Reset subscription to message position closest to given position.", notes = "It fence cursor and disconnects all active consumers before reseting cursor.")
+    @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
+            @ApiResponse(code = 404, message = "Topic/Subscription does not exist"),
+            @ApiResponse(code = 405, message = "Not supported for partitioned topics") })
+    public void resetCursorOnPosition(@PathParam("property") String property, @PathParam("cluster") String cluster,
+            @PathParam("namespace") String namespace, @PathParam("destination") @Encoded String destination,
+            @PathParam("subName") String subName,
+            @QueryParam("authoritative") @DefaultValue("false") boolean authoritative, MessageIdImpl messageId) {
+        destination = decode(destination);
+        DestinationName dn = DestinationName.get(domain(), property, cluster, namespace, destination);
+        log.info("[{}][{}] received reset cursor on subscription {} to position {}", clientAppId(), destination,
+                subName, messageId);
+
+        PartitionedTopicMetadata partitionMetadata = getPartitionedTopicMetadata(property, cluster, namespace,
+                destination, authoritative);
+
+        if (partitionMetadata.partitions > 0) {
+            log.warn("[{}] Not supported operation on partitioned-topic {} {}", clientAppId(), dn, subName);
+            throw new RestException(Status.METHOD_NOT_ALLOWED,
+                    "Reset-cursor at position is not allowed for partitioned-topic");
+        } else {
+            validateAdminOperationOnDestination(dn, authoritative);
+            PersistentTopic topic = (PersistentTopic) getTopicReference(dn);
+            if (topic == null) {
+                throw new RestException(Status.NOT_FOUND, "Topic not found");
+            }
+            try {
+                PersistentSubscription sub = topic.getSubscription(subName);
+                checkNotNull(sub);
+                sub.resetCursor(PositionImpl.get(messageId.getLedgerId(), messageId.getEntryId())).get();
+                log.info("[{}][{}] successfully reset cursor on subscription {} to position {}", clientAppId(), dn,
+                        subName, messageId);
+            } catch (Exception e) {
+                Throwable t = e.getCause();
+                log.warn("[{}] [{}] Failed to reset cursor on subscription {} to position {}", clientAppId(), dn,
+                        subName, messageId, e);
+                if (e instanceof NullPointerException) {
+                    throw new RestException(Status.NOT_FOUND, "Subscription not found");
+                } else if (t instanceof SubscriptionInvalidCursorPosition) {
+                    throw new RestException(Status.PRECONDITION_FAILED,
+                            "Unable to find position for position specified: " + t.getMessage());
+                } else {
+                    throw new RestException(e);
+                }
+            }
+        }
+    }
+    
     @GET
     @Path("/{property}/{cluster}/{namespace}/{destination}/subscription/{subName}/position/{messagePosition}")
     @ApiOperation(value = "Peek nth message on a topic subscription.")

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Subscription.java

@@ -22,8 +22,8 @@ import java.util.List;
 import java.util.concurrent.CompletableFuture;
 
 import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.Position;
 import org.apache.bookkeeper.mledger.impl.PositionImpl;
-import org.apache.pulsar.common.api.proto.PulsarApi;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandSubscribe.SubType;
 
@@ -59,6 +59,8 @@ public interface Subscription {
     CompletableFuture<Void> skipMessages(int numMessagesToSkip);
 
     CompletableFuture<Void> resetCursor(long timestamp);
+    
+    CompletableFuture<Void> resetCursor(Position position);
 
     CompletableFuture<Entry> peekNthMessage(int messagePosition);
 

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java

@@ -23,6 +23,7 @@ import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 
 import org.apache.bookkeeper.mledger.Entry;
+import org.apache.bookkeeper.mledger.Position;
 import org.apache.bookkeeper.mledger.impl.PositionImpl;
 import org.apache.pulsar.broker.service.BrokerServiceException;
 import org.apache.pulsar.broker.service.BrokerServiceException.ServerMetadataException;
@@ -347,6 +348,11 @@ public class NonPersistentSubscription implements Subscription {
         topic.markBatchMessagePublished();
     }
 
+    @Override
+    public CompletableFuture<Void> resetCursor(Position position) {
+        return CompletableFuture.completedFuture(null);
+    }
+
     private static final Logger log = LoggerFactory.getLogger(NonPersistentSubscription.class);
 
 }

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java

@@ -28,6 +28,7 @@ import org.apache.bookkeeper.mledger.AsyncCallbacks.CloseCallback;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.DeleteCallback;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.MarkDeleteCallback;
 import org.apache.bookkeeper.mledger.AsyncCallbacks.ReadEntryCallback;
+import org.apache.bookkeeper.mledger.AsyncCallbacks.ResetCursorCallback;
 import org.apache.bookkeeper.mledger.Entry;
 import org.apache.bookkeeper.mledger.ManagedCursor;
 import org.apache.bookkeeper.mledger.ManagedCursor.IndividualDeletedEntries;
@@ -343,62 +344,7 @@ public class PersistentSubscription implements Subscription {
                 } else {
                     finalPosition = position;
                 }
-
-                if (!IS_FENCED_UPDATER.compareAndSet(PersistentSubscription.this, FALSE, TRUE)) {
-                    future.completeExceptionally(new SubscriptionBusyException("Failed to fence subscription"));
-                    return;
-                }
-
-                final CompletableFuture<Void> disconnectFuture;
-                if (dispatcher != null && dispatcher.isConsumerConnected()) {
-                    disconnectFuture = dispatcher.disconnectAllConsumers();
-                } else {
-                    disconnectFuture = CompletableFuture.completedFuture(null);
-                }
-
-                disconnectFuture.whenComplete((aVoid, throwable) -> {
-                    if (throwable != null) {
-                        log.error("[{}][{}] Failed to disconnect consumer from subscription", topicName, subName, throwable);
-                        IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
-                        future.completeExceptionally(new SubscriptionBusyException("Failed to disconnect consumers from subscription"));
-                        return;
-                    }
-                    log.info("[{}][{}] Successfully disconnected consumers from subscription, proceeding with cursor reset", topicName, subName);
-
-                    try {
-                        cursor.asyncResetCursor(finalPosition, new AsyncCallbacks.ResetCursorCallback() {
-                            @Override
-                            public void resetComplete(Object ctx) {
-                                if (log.isDebugEnabled()) {
-                                    log.debug("[{}][{}] Successfully reset subscription to timestamp {}", topicName, subName,
-                                            timestamp);
-                                }
-                                IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
-                                future.complete(null);
-                            }
-
-                            @Override
-                            public void resetFailed(ManagedLedgerException exception, Object ctx) {
-                                log.error("[{}][{}] Failed to reset subscription to timestamp {}", topicName, subName, timestamp,
-                                        exception);
-                                IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
-                                // todo - retry on InvalidCursorPositionException
-                                // or should we just ask user to retry one more time?
-                                if (exception instanceof InvalidCursorPositionException) {
-                                    future.completeExceptionally(new SubscriptionInvalidCursorPosition(exception.getMessage()));
-                                } else if (exception instanceof ConcurrentFindCursorPositionException) {
-                                    future.completeExceptionally(new SubscriptionBusyException(exception.getMessage()));
-                                } else {
-                                    future.completeExceptionally(new BrokerServiceException(exception));
-                                }
-                            }
-                        });
-                    } catch (Exception e) {
-                        log.error("[{}][{}] Error while resetting cursor", topicName, subName, e);
-                        IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
-                        future.completeExceptionally(new BrokerServiceException(e));
-                    }
-                });
+                resetCursor(finalPosition, future);
             }
 
             @Override
@@ -415,6 +361,73 @@ public class PersistentSubscription implements Subscription {
         return future;
     }
 
+    @Override
+    public CompletableFuture<Void> resetCursor(Position position) {
+        CompletableFuture<Void> future = new CompletableFuture<>();
+        resetCursor(position, future);
+        return future;
+    }
+
+    private void resetCursor(Position finalPosition, CompletableFuture<Void> future) {
+        if (!IS_FENCED_UPDATER.compareAndSet(PersistentSubscription.this, FALSE, TRUE)) {
+            future.completeExceptionally(new SubscriptionBusyException("Failed to fence subscription"));
+            return;
+        }
+
+        final CompletableFuture<Void> disconnectFuture;
+        if (dispatcher != null && dispatcher.isConsumerConnected()) {
+            disconnectFuture = dispatcher.disconnectAllConsumers();
+        } else {
+            disconnectFuture = CompletableFuture.completedFuture(null);
+        }
+
+        disconnectFuture.whenComplete((aVoid, throwable) -> {
+            if (throwable != null) {
+                log.error("[{}][{}] Failed to disconnect consumer from subscription", topicName, subName, throwable);
+                IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
+                future.completeExceptionally(
+                        new SubscriptionBusyException("Failed to disconnect consumers from subscription"));
+                return;
+            }
+            log.info("[{}][{}] Successfully disconnected consumers from subscription, proceeding with cursor reset",
+                    topicName, subName);
+
+            try {
+                cursor.asyncResetCursor(finalPosition, new AsyncCallbacks.ResetCursorCallback() {
+                    @Override
+                    public void resetComplete(Object ctx) {
+                        if (log.isDebugEnabled()) {
+                            log.debug("[{}][{}] Successfully reset subscription to position {}", topicName, subName,
+                                    finalPosition);
+                        }
+                        IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
+                        future.complete(null);
+                    }
+
+                    @Override
+                    public void resetFailed(ManagedLedgerException exception, Object ctx) {
+                        log.error("[{}][{}] Failed to reset subscription to position {}", topicName, subName,
+                                finalPosition, exception);
+                        IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
+                        // todo - retry on InvalidCursorPositionException
+                        // or should we just ask user to retry one more time?
+                        if (exception instanceof InvalidCursorPositionException) {
+                            future.completeExceptionally(new SubscriptionInvalidCursorPosition(exception.getMessage()));
+                        } else if (exception instanceof ConcurrentFindCursorPositionException) {
+                            future.completeExceptionally(new SubscriptionBusyException(exception.getMessage()));
+                        } else {
+                            future.completeExceptionally(new BrokerServiceException(exception));
+                        }
+                    }
+                });
+            } catch (Exception e) {
+                log.error("[{}][{}] Error while resetting cursor", topicName, subName, e);
+                IS_FENCED_UPDATER.set(PersistentSubscription.this, FALSE);
+                future.completeExceptionally(new BrokerServiceException(e));
+            }
+        });
+    }
+
     @Override
     public CompletableFuture<Entry> peekNthMessage(int messagePosition) {
         CompletableFuture<Entry> future = new CompletableFuture<>();

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest.java

@@ -21,8 +21,6 @@ package org.apache.pulsar.broker.admin;
 import static org.testng.Assert.assertEquals;
 import static org.testng.Assert.assertFalse;
 import static org.testng.Assert.assertNotEquals;
-import static org.testng.Assert.assertNotNull;
-import static org.testng.Assert.assertNull;
 import static org.testng.Assert.assertTrue;
 import static org.testng.Assert.fail;
 
@@ -48,8 +46,6 @@ import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
 import org.apache.pulsar.broker.namespace.NamespaceEphemeralData;
 import org.apache.pulsar.broker.namespace.NamespaceService;
 import org.apache.pulsar.broker.service.BrokerService;
-import org.apache.pulsar.broker.service.Topic;
-import org.apache.pulsar.broker.service.persistent.PersistentTopic;
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.admin.PulsarAdminException.ConflictException;
@@ -86,7 +82,6 @@ import org.apache.pulsar.common.policies.data.BrokerAssignment;
 import org.apache.pulsar.common.policies.data.ClusterData;
 import org.apache.pulsar.common.policies.data.NamespaceIsolationData;
 import org.apache.pulsar.common.policies.data.NamespaceOwnershipStatus;
-import org.apache.pulsar.common.policies.data.NonPersistentTopicStats;
 import org.apache.pulsar.common.policies.data.PartitionedTopicStats;
 import org.apache.pulsar.common.policies.data.PersistencePolicies;
 import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
@@ -1753,57 +1748,4 @@ public class AdminApiTest extends MockedPulsarServiceBaseTest {
                 pulsar.getNamespaceService().getBundle(DestinationName.get(topicName)).getBundleRange());
     }
 
-    // TODO: move to AdminApiTest2.java
-    /**
-     * Verify unloading topic
-     * 
-     * @throws Exception
-     */
-    @Test(dataProvider = "topicType")
-    public void testUnloadTopic(final String topicType) throws Exception {
-
-        final String namespace = "prop-xyz/use/ns2";
-        final String topicName = topicType + "://" + namespace + "/topic1";
-        admin.namespaces().createNamespace(namespace);
-
-        // create a topic by creating a producer
-        Producer producer = pulsarClient.createProducer(topicName);
-        producer.close();
-
-        Topic topic = pulsar.getBrokerService().getTopicReference(topicName);
-        assertNotNull(topic);
-        final boolean isPersistentTopic = topic instanceof PersistentTopic;
-
-        // (1) unload the topic
-        unloadTopic(topicName, isPersistentTopic);
-        topic = pulsar.getBrokerService().getTopicReference(topicName);
-        // topic must be removed
-        assertNull(topic);
-
-        // recreation of producer will load the topic again
-        producer = pulsarClient.createProducer(topicName);
-        topic = pulsar.getBrokerService().getTopicReference(topicName);
-        assertNotNull(topic);
-        // unload the topic
-        unloadTopic(topicName, isPersistentTopic);
-        // producer will retry and recreate the topic
-        for (int i = 0; i < 5; i++) {
-            topic = pulsar.getBrokerService().getTopicReference(topicName);
-            if (topic == null || i != 4) {
-                Thread.sleep(200);
-            }
-        }
-        // topic should be loaded by this time
-        topic = pulsar.getBrokerService().getTopicReference(topicName);
-        assertNotNull(topic);
-    }
-
-    private void unloadTopic(String topicName, boolean isPersistentTopic) throws Exception {
-        if (isPersistentTopic) {
-            admin.persistentTopics().unload(topicName);
-        } else {
-            admin.nonPersistentTopics().unload(topicName);
-        }
-    }
-
 }

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java

@@ -19,10 +19,13 @@
 package org.apache.pulsar.broker.admin;
 
 import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNull;
 import static org.testng.Assert.assertTrue;
 import static org.testng.Assert.fail;
 
 import java.net.URL;
+import java.util.List;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
@@ -31,17 +34,21 @@ import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
 import org.apache.pulsar.broker.PulsarService;
 import org.apache.pulsar.broker.admin.AdminApiTest.MockedPulsarService;
 import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
+import org.apache.pulsar.broker.service.Topic;
 import org.apache.pulsar.broker.service.persistent.PersistentTopic;
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.api.ClientConfiguration;
 import org.apache.pulsar.client.api.Consumer;
 import org.apache.pulsar.client.api.ConsumerConfiguration;
+import org.apache.pulsar.client.api.Message;
 import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.ProducerConfiguration;
+import org.apache.pulsar.client.api.ProducerConfiguration.MessageRoutingMode;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.SubscriptionType;
-import org.apache.pulsar.client.api.ProducerConfiguration.MessageRoutingMode;
+import org.apache.pulsar.client.impl.MessageIdImpl;
+import org.apache.pulsar.common.naming.DestinationDomain;
 import org.apache.pulsar.common.naming.DestinationName;
 import org.apache.pulsar.common.naming.NamespaceBundleFactory;
 import org.apache.pulsar.common.policies.data.ClusterData;
@@ -51,10 +58,12 @@ import org.apache.pulsar.common.policies.data.PersistencePolicies;
 import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
 import org.apache.pulsar.common.policies.data.PersistentTopicStats;
 import org.apache.pulsar.common.policies.data.PropertyAdmin;
+import org.apache.pulsar.common.policies.data.RetentionPolicies;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
 import com.google.common.collect.Lists;
@@ -103,6 +112,17 @@ public class AdminApiTest2 extends MockedPulsarServiceBaseTest {
         mockPulsarSetup.cleanup();
     }
 
+    @DataProvider(name = "topicType")
+    public Object[][] topicTypeProvider() {
+        return new Object[][] { { DestinationDomain.persistent.value() },
+                { DestinationDomain.non_persistent.value() } };
+    }
+
+    @DataProvider(name = "namespaceNames")
+    public Object[][] namespaceNameProvider() {
+        return new Object[][] { { "ns1" }, { "global" } };
+    }
+
     /**
      * <pre>
      * It verifies increasing partitions for partitioned-topic.
@@ -345,4 +365,163 @@ public class AdminApiTest2 extends MockedPulsarServiceBaseTest {
 
     }
 
+    /**
+     * Verify unloading topic
+     * 
+     * @throws Exception
+     */
+    @Test(dataProvider = "topicType")
+    public void testUnloadTopic(final String topicType) throws Exception {
+
+        final String namespace = "prop-xyz/use/ns2";
+        final String topicName = topicType + "://" + namespace + "/topic1";
+        admin.namespaces().createNamespace(namespace);
+
+        // create a topic by creating a producer
+        Producer producer = pulsarClient.createProducer(topicName);
+        producer.close();
+
+        Topic topic = pulsar.getBrokerService().getTopicReference(topicName);
+        assertNotNull(topic);
+        final boolean isPersistentTopic = topic instanceof PersistentTopic;
+
+        // (1) unload the topic
+        unloadTopic(topicName, isPersistentTopic);
+        topic = pulsar.getBrokerService().getTopicReference(topicName);
+        // topic must be removed
+        assertNull(topic);
+
+        // recreation of producer will load the topic again
+        producer = pulsarClient.createProducer(topicName);
+        topic = pulsar.getBrokerService().getTopicReference(topicName);
+        assertNotNull(topic);
+        // unload the topic
+        unloadTopic(topicName, isPersistentTopic);
+        // producer will retry and recreate the topic
+        for (int i = 0; i < 5; i++) {
+            topic = pulsar.getBrokerService().getTopicReference(topicName);
+            if (topic == null || i != 4) {
+                Thread.sleep(200);
+            }
+        }
+        // topic should be loaded by this time
+        topic = pulsar.getBrokerService().getTopicReference(topicName);
+        assertNotNull(topic);
+    }
+
+    private void unloadTopic(String topicName, boolean isPersistentTopic) throws Exception {
+        if (isPersistentTopic) {
+            admin.persistentTopics().unload(topicName);
+        } else {
+            admin.nonPersistentTopics().unload(topicName);
+        }
+    }
+
+    /**
+     * Verifies reset-cursor at specific position using admin-api.
+     * 
+     * <pre>
+     * 1. Publish 50 messages
+     * 2. Consume 20 messages
+     * 3. reset cursor position on 10th message
+     * 4. consume 40 messages from reset position
+     * </pre>
+     * 
+     * @param namespaceName
+     * @throws Exception
+     */
+    @Test(dataProvider = "namespaceNames", timeOut = 10000)
+    public void testResetCursorOnPosition(String namespaceName) throws Exception {
+        final String topicName = "persistent://prop-xyz/use/" + namespaceName + "/resetPosition";
+        final int totalProducedMessages = 50;
+
+        // set retention
+        admin.namespaces().setRetention("prop-xyz/use/ns1", new RetentionPolicies(10, 10));
+
+        // create consumer and subscription
+        ConsumerConfiguration conf = new ConsumerConfiguration();
+        conf.setSubscriptionType(SubscriptionType.Shared);
+        Consumer consumer = pulsarClient.subscribe(topicName, "my-sub", conf);
+
+        assertEquals(admin.persistentTopics().getSubscriptions(topicName), Lists.newArrayList("my-sub"));
+
+        publishMessagesOnPersistentTopic(topicName, totalProducedMessages, 0);
+
+        List<Message> messages = admin.persistentTopics().peekMessages(topicName, "my-sub", 10);
+        assertEquals(messages.size(), 10);
+
+        Message message = null;
+        MessageIdImpl resetMessageId = null;
+        int resetPositionId = 10;
+        for (int i = 0; i < 20; i++) {
+            message = consumer.receive(1, TimeUnit.SECONDS);
+            consumer.acknowledge(message);
+            if (i == resetPositionId) {
+                resetMessageId = (MessageIdImpl) message.getMessageId();
+            }
+        }
+
+        // close consumer which will clean up intenral-receive-queue
+        consumer.close();
+
+        // messages should still be available due to retention
+        MessageIdImpl messageId = new MessageIdImpl(resetMessageId.getLedgerId(), resetMessageId.getEntryId(), -1);
+        // reset position at resetMessageId
+        admin.persistentTopics().resetCursor(topicName, "my-sub", messageId);
+
+        consumer = pulsarClient.subscribe(topicName, "my-sub", conf);
+        MessageIdImpl msgId2 = (MessageIdImpl) consumer.receive(1, TimeUnit.SECONDS).getMessageId();
+        assertEquals(resetMessageId, msgId2);
+
+        int receivedAfterReset = 1; // start with 1 because we have already received 1 msg
+
+        for (int i = 0; i < totalProducedMessages; i++) {
+            message = consumer.receive(500, TimeUnit.MILLISECONDS);
+            if (message == null) {
+                break;
+            }
+            consumer.acknowledge(message);
+            ++receivedAfterReset;
+        }
+        assertEquals(receivedAfterReset, totalProducedMessages - resetPositionId);
+
+        // invalid topic name
+        try {
+            admin.persistentTopics().resetCursor(topicName + "invalid", "my-sub", messageId);
+            fail("It should have failed due to invalid topic name");
+        } catch (PulsarAdminException.NotFoundException e) {
+            // Ok
+        }
+
+        // invalid cursor name
+        try {
+            admin.persistentTopics().resetCursor(topicName, "invalid-sub", messageId);
+            fail("It should have failed due to invalid subscription name");
+        } catch (PulsarAdminException.NotFoundException e) {
+            // Ok
+        }
+
+        // invalid position
+        try {
+            messageId = new MessageIdImpl(0, 0, -1);
+            admin.persistentTopics().resetCursor(topicName, "my-sub", messageId);
+            fail("It should have failed due to invalid subscription name");
+        } catch (PulsarAdminException.PreconditionFailedException e) {
+            // Ok
+        }
+
+        consumer.close();
+    }
+
+    private void publishMessagesOnPersistentTopic(String topicName, int messages, int startIdx) throws Exception {
+        Producer producer = pulsarClient.createProducer(topicName);
+
+        for (int i = startIdx; i < (messages + startIdx); i++) {
+            String message = "message-" + i;
+            producer.send(message.getBytes());
+        }
+
+        producer.close();
+    }
+
 }

pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/PersistentTopics.java

@@ -829,4 +829,37 @@ public interface PersistentTopics {
      *            reset subscription to position closest to time in ms since epoch
      */
     CompletableFuture<Void> resetCursorAsync(String destination, String subName, long timestamp);
+    
+    /**
+     * Reset cursor position on a topic subscription
+     *
+     * @param destination
+     *            Destination name
+     * @param subName
+     *            Subscription name
+     * @param messageId
+     *            reset subscription to messageId (or previous nearest messageId if given messageId is not valid)
+     *
+     * @throws NotAuthorizedException
+     *             Don't have admin permission
+     * @throws NotFoundException
+     *             Topic or subscription does not exist
+     * @throws NotAllowedException
+     *             Command disallowed for requested resource
+     * @throws PulsarAdminException
+     *             Unexpected error
+     */
+    void resetCursor(String destination, String subName, MessageId messageId) throws PulsarAdminException;
+
+    /**
+     * Reset cursor position on a topic subscription
+     *
+     * @param destination
+     *            Destination name
+     * @param subName
+     *            Subscription name
+     * @param MessageId
+     *            reset subscription to messageId (or previous nearest messageId if given messageId is not valid)
+     */
+    CompletableFuture<Void> resetCursorAsync(String destination, String subName, MessageId messageId);
 }

pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/PersistentTopicsImpl.java

@@ -640,6 +640,28 @@ public class PersistentTopicsImpl extends BaseResource implements PersistentTopi
                 Entity.entity("", MediaType.APPLICATION_JSON));
     }
 
+    @Override
+    public void resetCursor(String destination, String subName, MessageId messageId) throws PulsarAdminException {
+        try {
+            DestinationName ds = validateTopic(destination);
+            String encodedSubName = Codec.encode(subName);
+            request(persistentTopics.path(ds.getNamespace()).path(ds.getEncodedLocalName()).path("subscription")
+                    .path(encodedSubName).path("resetcursor")).post(Entity.entity(messageId, MediaType.APPLICATION_JSON),
+                            ErrorData.class);
+        } catch (Exception e) {
+            throw getApiException(e);
+        }
+    }
+
+    @Override
+    public CompletableFuture<Void> resetCursorAsync(String destination, String subName, MessageId messageId) {
+        DestinationName ds = validateTopic(destination);
+        String encodedSubName = Codec.encode(subName);
+        return asyncPostRequest(persistentTopics.path(ds.getNamespace()).path(ds.getEncodedLocalName())
+                .path("subscription").path(encodedSubName).path("resetcursor"),
+                Entity.entity(messageId, MediaType.APPLICATION_JSON));
+    }
+
     @Override
     public CompletableFuture<MessageId> terminateTopicAsync(String destination) {
         DestinationName ds = validateTopic(destination);

pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdPersistentTopics.java

@@ -22,6 +22,7 @@ import java.util.List;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
 
+import static org.apache.commons.lang3.StringUtils.isNotBlank;
 import org.apache.pulsar.client.admin.PersistentTopics;
 import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.admin.PulsarAdminException;
@@ -33,6 +34,7 @@ import org.apache.pulsar.client.impl.MessageIdImpl;
 import com.beust.jcommander.Parameter;
 import com.beust.jcommander.Parameters;
 import com.beust.jcommander.converters.CommaParameterSplitter;
+import static com.google.common.base.Preconditions.checkArgument;
 import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
 import com.google.gson.JsonObject;
@@ -416,7 +418,7 @@ public class CmdPersistentTopics extends CmdBase {
         }
     }
 
-    @Parameters(commandDescription = "Reset position for subscription to position closest to timestamp")
+    @Parameters(commandDescription = "Reset position for subscription to position closest to timestamp or messageId")
     private class ResetCursor extends CliCommand {
         @Parameter(description = "persistent://property/cluster/namespace/destination", required = true)
         private java.util.List<String> params;
@@ -426,17 +428,29 @@ public class CmdPersistentTopics extends CmdBase {
         private String subName;
 
         @Parameter(names = { "--time",
-                "-t" }, description = "time in minutes to reset back to (or minutes, hours,days,weeks eg: 100m, 3h, 2d, 5w)", required = true)
+                "-t" }, description = "time in minutes to reset back to (or minutes, hours,days,weeks eg: 100m, 3h, 2d, 5w)", required = false)
         private String resetTimeStr;
+        
+        @Parameter(names = { "--messageId",
+                "-m" }, description = "messageId to reset back to (ledgerId:entryId)", required = false)
+        private String resetMessageIdStr;
 
         @Override
         void run() throws PulsarAdminException {
             String persistentTopic = validatePersistentTopic(params);
-            int resetBackTimeInMin = validateTimeString(resetTimeStr);
-            long resetTimeInMillis = TimeUnit.MILLISECONDS.convert(resetBackTimeInMin, TimeUnit.MINUTES);
-            // now - go back time
-            long timestamp = System.currentTimeMillis() - resetTimeInMillis;
-            persistentTopics.resetCursor(persistentTopic, subName, timestamp);
+            if (isNotBlank(resetMessageIdStr)) {
+                MessageId messageId = validateMessageIdString(resetMessageIdStr);
+                persistentTopics.resetCursor(persistentTopic, subName, messageId);
+            } else if (isNotBlank(resetTimeStr)) {
+                int resetBackTimeInMin = validateTimeString(resetTimeStr);
+                long resetTimeInMillis = TimeUnit.MILLISECONDS.convert(resetBackTimeInMin, TimeUnit.MINUTES);
+                // now - go back time
+                long timestamp = System.currentTimeMillis() - resetTimeInMillis;
+                persistentTopics.resetCursor(persistentTopic, subName, timestamp);
+            } else {
+                throw new PulsarAdminException(
+                        "Either Timestamp (--time) or Position (--position) has to be provided to reset cursor");
+            }
         }
     }
 
@@ -520,4 +534,15 @@ public class CmdPersistentTopics extends CmdBase {
             return Integer.parseInt(s);
         }
     }
+    
+    private MessageId validateMessageIdString(String resetMessageIdStr) throws PulsarAdminException {
+        String[] messageId = resetMessageIdStr.split(":");
+        try {
+            checkArgument(messageId.length == 2);
+            return new MessageIdImpl(Long.parseLong(messageId[0]), Long.parseLong(messageId[1]), -1);
+        } catch (Exception e) {
+            throw new PulsarAdminException(
+                    "Invalid reset-position (must be in format: ledgerId:entryId) value " + resetMessageIdStr);
+        }
+    }
 }