Add Cli

Author: xiangying

pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/PulsarAdminToolTest.java

@@ -2090,7 +2090,8 @@ public boolean matches(Long timestamp) {
         cmdTopics.run(split("get-shadow-source persistent://myprop/clust/ns1/ds1"));
         verify(mockTopics).getShadowSource("persistent://myprop/clust/ns1/ds1");
 
-
+        cmdTopics.run(split("get-message-id-by-offset persistent://myprop/clust/ns1/ds1 -o 0"));
+        verify(mockTopics).getMessageIdByOffset("persistent://myprop/clust/ns1/ds1", 0);
 
     }
 

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

@@ -275,6 +275,8 @@ public CmdTopics(Supplier<PulsarAdmin> admin) {
         addCommand("set-schema-validation-enforce", new SetSchemaValidationEnforced());
 
         addCommand("trim-topic", new TrimTopic());
+
+        addCommand("get-message-id-by-offset", new GetMessageIdByOffset());
     }
 
     @Command(description = "Get the list of topics under a namespace.")
@@ -3058,4 +3060,20 @@ void run() throws PulsarAdminException {
             getAdmin().topics().trimTopic(topic);
         }
     }
+
+    @Command(description = "Get message id by offset")
+    private class GetMessageIdByOffset extends CliCommand {
+
+        @Parameters(description = "persistent://tenant/namespace/topic", arity = "1")
+        private String topicName;
+
+        @Option(names = { "--offset", "-o" }, description = "Offset to get message id for the topic", required = true)
+        private Long offset;
+
+        @Override
+        void run() throws Exception {
+            String topic = validateTopicName(topicName);
+            getAdmin().topics().getMessageIdByOffset(topic, offset);
+        }
+    }
 }