[fix][cli] Add get-cluster-migration cmd (apache#21473)

liudezhi2098 · Dec 6, 2023 · ab77ca2 · ab77ca2
1 parent 1919a0e
commit ab77ca2
Show file tree

Hide file tree

Showing 3 changed files with 27 additions and 17 deletions.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ClustersBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ClustersBase.java
@@ -61,7 +61,6 @@
 import org.apache.pulsar.common.policies.data.BrokerNamespaceIsolationDataImpl;
 import org.apache.pulsar.common.policies.data.ClusterData;
 import org.apache.pulsar.common.policies.data.ClusterDataImpl;
-import org.apache.pulsar.common.policies.data.ClusterPolicies;
 import org.apache.pulsar.common.policies.data.ClusterPolicies.ClusterUrl;
 import org.apache.pulsar.common.policies.data.ClusterPoliciesImpl;
 import org.apache.pulsar.common.policies.data.FailureDomainImpl;
@@ -262,26 +261,29 @@ public void updateCluster(
             @ApiResponse(code = 404, message = "Cluster doesn't exist."),
             @ApiResponse(code = 500, message = "Internal server error.")
     })
-    public ClusterPolicies getClusterMigration(
+    public void getClusterMigration(
+        @Suspended AsyncResponse asyncResponse,
         @ApiParam(
             value = "The cluster name",
             required = true
         )
-        @PathParam("cluster") String cluster
-    ) {
-        validateSuperUserAccess();
-
-        try {
-            return clusterResources().getClusterPoliciesResources().getClusterPolicies(cluster)
-                    .orElseThrow(() -> new RestException(Status.NOT_FOUND, "Cluster does not exist"));
-        } catch (Exception e) {
-            log.error("[{}] Failed to get cluster {}", clientAppId(), cluster, e);
-            if (e instanceof RestException) {
-                throw (RestException) e;
-            } else {
-                throw new RestException(e);
-            }
-        }
+        @PathParam("cluster") String cluster) {
+        validateSuperUserAccessAsync()
+                .thenCompose(__ -> clusterResources().getClusterPoliciesResources().getClusterPoliciesAsync(cluster))
+                .thenAccept(policies -> {
+                    asyncResponse.resume(
+                            policies.orElseThrow(() -> new RestException(Status.NOT_FOUND, "Cluster does not exist")));
+                })
+                .exceptionally(ex -> {
+                    log.error("[{}] Failed to get cluster {} migration", clientAppId(), cluster, ex);
+                    Throwable realCause = FutureUtil.unwrapCompletionException(ex);
+                    if (realCause instanceof MetadataStoreException.NotFoundException) {
+                        asyncResponse.resume(new RestException(Status.NOT_FOUND, "Cluster does not exist"));
+                        return null;
+                    }
+                    resumeAsyncResponseExceptionally(asyncResponse, ex);
+                    return null;
+                });
     }
 
     @POST

diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdClusters.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdClusters.java
@@ -485,6 +485,7 @@ public CmdClusters(Supplier<PulsarAdmin> admin) {
         jcommander.addCommand("delete", new Delete());
         jcommander.addCommand("list", new List());
         jcommander.addCommand("update-peer-clusters", new UpdatePeerClusters());
+        jcommander.addCommand("get-cluster-migration", new GetClusterMigration());
         jcommander.addCommand("update-cluster-migration", new UpdateClusterMigration());
         jcommander.addCommand("get-peer-clusters", new GetPeerClusters());
         jcommander.addCommand("get-failure-domain", new GetFailureDomain());

diff --git a/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdClusters.java b/pulsar-client-tools/src/test/java/org/apache/pulsar/admin/cli/TestCmdClusters.java
@@ -25,6 +25,7 @@
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 import java.io.ByteArrayOutputStream;
@@ -122,4 +123,10 @@ public void testListCmd() throws Exception {
             System.setOut(defaultSystemOut);
         }
     }
+
+    @Test
+    public void testGetClusterMigration() throws Exception {
+        cmdClusters.run(new String[]{"get-cluster-migration", "test_cluster"});
+        verify(clusters, times(1)).getClusterMigration("test_cluster");
+    }
 }