[pulsar] branch master updated: cli support for get last message id of a topic (#3251)

[sijie]

This is an automated email from the ASF dual-hosted git repository.

sijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git


The following commit(s) were added to refs/heads/master by this push:
     new 94a9423  cli support for get last message id of a topic (#3251)
94a9423 is described below

commit 94a942384b4d1e07fba4dfb1fe2cc0c06bc0a2ea
Author: legendtkl <taok...@gmail.com>
AuthorDate: Thu Dec 27 00:14:50 2018 +0800

    cli support for get last message id of a topic (#3251)
    
    This pr is add cli support to get last message id of a topic
    1. add cli support for api: 
/admin/v2/persistent/{tenant}/{namespace}/{topic}/lastMessageId
    2. the cli command is like: ./pulsar-admin topics last-message-id topic-name
    
    Test Result:
    
    ```bash
    ➜  bin ./pulsar-admin topics last-message-id 
persistent://public/default/my-topic
    {
      "ledgerId" : 10,
      "entryId" : -1,
      "partitionIndex" : -1
    }
    ➜  bin ./pulsar-admin topics last-message-id 
persistent://public/default/my-topi
    Topic not found
    
    Reason: Topic not found
    ```
---
 .../org/apache/pulsar/client/admin/Topics.java     |  9 ++++++
 .../pulsar/client/admin/internal/TopicsImpl.java   | 32 ++++++++++++++++++++++
 .../org/apache/pulsar/admin/cli/CmdTopics.java     | 13 +++++++++
 3 files changed, 54 insertions(+)

diff --git 
a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Topics.java 
b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Topics.java
index f3f95ac..9c442c7 100644
--- 
a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Topics.java
+++ 
b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/Topics.java
@@ -1017,4 +1017,13 @@ public interface Topics {
      * @return the status of the offload operation
      */
     OffloadProcessStatus offloadStatus(String topic) throws 
PulsarAdminException;
+
+    /**
+     * Get the last commit message Id of a topic
+     *
+     * @param topic the topic name
+     * @return
+     * @throws PulsarAdminException
+     */
+    MessageId getLastMessageId(String topic) throws PulsarAdminException;
 }
diff --git 
a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java
 
b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java
index 216275d..5dc84fc 100644
--- 
a/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java
+++ 
b/pulsar-client-admin/src/main/java/org/apache/pulsar/client/admin/internal/TopicsImpl.java
@@ -950,5 +950,37 @@ public class TopicsImpl extends BaseResource implements 
Topics, PersistentTopics
         return ret;
     }
 
+    @Override
+    public MessageId getLastMessageId(String topic) throws 
PulsarAdminException {
+        try {
+            return (MessageIdImpl) getLastMessageIdAsync(topic).get();
+        } catch (ExecutionException e) {
+            throw (PulsarAdminException) e.getCause();
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new PulsarAdminException(e.getCause());
+        }
+    }
+
+    public CompletableFuture<MessageId> getLastMessageIdAsync(String topic) {
+        TopicName tn = validateTopic(topic);
+        WebTarget path = topicPath(tn, "lastMessageId");
+        final CompletableFuture<MessageId> future = new CompletableFuture<>();
+        asyncGetRequest(path,
+                new InvocationCallback<MessageIdImpl>() {
+
+                    @Override
+                    public void completed(MessageIdImpl response) {
+                        future.complete(response);
+                    }
+
+                    @Override
+                    public void failed(Throwable throwable) {
+                        
future.completeExceptionally(getApiException(throwable.getCause()));
+                    }
+                });
+        return future;
+    }
+
     private static final Logger log = 
LoggerFactory.getLogger(TopicsImpl.class);
 }
diff --git 
a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java 
b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
index 5014f14..ceaebf4 100644
--- 
a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
+++ 
b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdTopics.java
@@ -93,6 +93,7 @@ public class CmdTopics extends CmdBase {
         jcommander.addCommand("compaction-status", new CompactionStatusCmd());
         jcommander.addCommand("offload", new Offload());
         jcommander.addCommand("offload-status", new OffloadStatusCmd());
+        jcommander.addCommand("last-message-id", new GetLastMessageId());
     }
 
     @Parameters(commandDescription = "Get the list of topics under a 
namespace.")
@@ -708,4 +709,16 @@ public class CmdTopics extends CmdBase {
             }
         }
     }
+
+    @Parameters(commandDescription = "get the last commit message id of topic")
+    private class GetLastMessageId extends CliCommand {
+        @Parameter(description = "persistent://tenant/namespace/topic", 
required = true)
+        private java.util.List<String> params;
+
+        @Override
+        void run() throws PulsarAdminException {
+            String persistentTopic = validatePersistentTopic(params);
+            print(topics.getLastMessageId(persistentTopic));
+        }
+    }
 }