[INLONG-8674][Manager] Pulsar - Modify the calling method from SDK to HTTP

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/cluster/PulsarClusterOperator.java

@@ -133,7 +133,7 @@ private Boolean testConnectAdminUrl(PulsarClusterInfo pulsarInfo) {
 
         try {
             // test connect for pulsar adminUrl
-            PulsarUtils.getPulsarTenants(restTemplate, pulsarInfo);
+            PulsarUtils.getTenants(restTemplate, pulsarInfo);
             return true;
         } catch (Exception e) {
             String errMsg = String.format("Pulsar connection failed for AdminUrl=%s", pulsarInfo.getAdminUrl());

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/node/pulsar/PulsarDataNodeOperator.java

@@ -111,7 +111,7 @@ private boolean getPulsarConnection(String adminUrl, String token) {
                 .token(token).build();
         try {
             // test connect for pulsar adminUrl
-            PulsarUtils.getPulsarTenants(restTemplate, pulsarClusterInfo);
+            PulsarUtils.getTenants(restTemplate, pulsarClusterInfo);
             return true;
         } catch (Exception e) {
             String errMsg = String.format("Pulsar connection failed for AdminUrl=%s", pulsarClusterInfo.getAdminUrl());

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarOperator.java

@@ -80,7 +80,7 @@ public void createTenant(PulsarClusterInfo pulsarClusterInfo, String tenant) thr
         Preconditions.expectNotBlank(tenant, ErrorCodeEnum.INVALID_PARAMETER, "Tenant cannot be empty");
 
         try {
-            List<String> clusters = PulsarUtils.getPulsarClusters(restTemplate, pulsarClusterInfo);
+            List<String> clusters = PulsarUtils.getClusters(restTemplate, pulsarClusterInfo);
             boolean exists = this.tenantIsExists(pulsarClusterInfo, tenant);
             if (exists) {
                 LOGGER.warn("pulsar tenant={} already exists, skip to create", tenant);
@@ -179,10 +179,10 @@ public void createTopic(PulsarClusterInfo pulsarClusterInfo, PulsarTopicInfo top
                 LOGGER.info("success to create topic={}, lookup result is {}", fullTopicName, res);
             } else {
                 // The number of brokers as the default value of topic partition
-                List<String> clusters = PulsarUtils.getPulsarClusters(restTemplate, pulsarClusterInfo);
+                List<String> clusters = PulsarUtils.getClusters(restTemplate, pulsarClusterInfo);
                 Integer numPartitions = topicInfo.getNumPartitions();
                 if (numPartitions < 0 || numPartitions >= MAX_PARTITION) {
-                    List<String> brokers = PulsarUtils.getPulsarBrokers(restTemplate, pulsarClusterInfo);
+                    List<String> brokers = PulsarUtils.getBrokers(restTemplate, pulsarClusterInfo);
                     numPartitions = brokers.size();
                 }
                 PulsarUtils.createPartitionedTopic(restTemplate, pulsarClusterInfo, fullTopicName,
@@ -284,7 +284,7 @@ public void createSubscriptions(PulsarClusterInfo pulsarClusterInfo, String subs
      * @throws Exception any exception if occurred
      */
     private boolean tenantIsExists(PulsarClusterInfo pulsarClusterInfo, String tenant) throws Exception {
-        List<String> tenants = PulsarUtils.getPulsarTenants(restTemplate, pulsarClusterInfo);
+        List<String> tenants = PulsarUtils.getTenants(restTemplate, pulsarClusterInfo);
         return tenants.contains(tenant);
     }
 
@@ -299,7 +299,7 @@ private boolean tenantIsExists(PulsarClusterInfo pulsarClusterInfo, String tenan
      */
     private boolean namespaceExists(PulsarClusterInfo pulsarClusterInfo, String tenant, String namespace)
             throws Exception {
-        List<String> namespaces = PulsarUtils.getPulsarNamespaces(restTemplate, pulsarClusterInfo, tenant);
+        List<String> namespaces = PulsarUtils.getNamespaces(restTemplate, pulsarClusterInfo, tenant);
         return namespaces.contains(namespace);
     }
 
@@ -320,10 +320,10 @@ public boolean topicExists(PulsarClusterInfo pulsarClusterInfo, String tenant, S
         boolean topicExists = false;
         try {
             if (isPartitioned) {
-                topics = PulsarUtils.getPulsarPartitionedTopics(restTemplate, pulsarClusterInfo, tenant,
+                topics = PulsarUtils.getPartitionedTopics(restTemplate, pulsarClusterInfo, tenant,
                         namespace);
             } else {
-                topics = PulsarUtils.getPulsarTopics(restTemplate, pulsarClusterInfo, tenant, namespace);
+                topics = PulsarUtils.getTopics(restTemplate, pulsarClusterInfo, tenant, namespace);
             }
             for (String t : topics) {
                 t = t.substring(t.lastIndexOf("/") + 1); // not contains /
@@ -346,7 +346,7 @@ public boolean topicExists(PulsarClusterInfo pulsarClusterInfo, String tenant, S
                     LOGGER.info("check whether the pulsar topic={} exists error, try count={}", topicName, count);
                     Thread.sleep(DELAY_SECONDS);
 
-                    topics = PulsarUtils.getPulsarPartitionedTopics(restTemplate, pulsarClusterInfo,
+                    topics = PulsarUtils.getPartitionedTopics(restTemplate, pulsarClusterInfo,
                             tenant, namespace);
                     for (String t : topics) {
                         t = t.substring(t.lastIndexOf("/") + 1);
@@ -430,7 +430,7 @@ public List<BriefMQMessage> queryLatestMessage(PulsarClusterInfo pulsarClusterIn
     private int getPartitionCount(PulsarClusterInfo pulsarClusterInfo, String topicFullName) {
         PulsarTopicMetadata pulsarTopicMetadata;
         try {
-            pulsarTopicMetadata = PulsarUtils.getPulsarPartitionedTopicMetadata(restTemplate,
+            pulsarTopicMetadata = PulsarUtils.getPartitionedTopicMetadata(restTemplate,
                     pulsarClusterInfo, topicFullName);
         } catch (Exception e) {
             String errMsg = "get pulsar partition error ";

inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/resource/queue/pulsar/PulsarUtils.java

@@ -99,7 +99,7 @@ private static HttpHeaders getHttpHeaders(String token) {
      * @return list of pulsar cluster infos
      * @throws Exception any exception if occurred
      */
-    public static List<String> getPulsarClusters(RestTemplate restTemplate, PulsarClusterInfo clusterInfo)
+    public static List<String> getClusters(RestTemplate restTemplate, PulsarClusterInfo clusterInfo)
             throws Exception {
         final String url = clusterInfo.getAdminUrl() + QUERY_CLUSTERS_PATH;
         return HttpUtils.request(restTemplate, url, HttpMethod.GET, null, getHttpHeaders(clusterInfo.getToken()),
@@ -114,17 +114,17 @@ public static List<String> getPulsarClusters(RestTemplate restTemplate, PulsarCl
      * @return list of pulsar broker infos
      * @throws Exception any exception if occurred
      */
-    public static List<String> getPulsarBrokers(RestTemplate restTemplate, PulsarClusterInfo clusterInfo)
+    public static List<String> getBrokers(RestTemplate restTemplate, PulsarClusterInfo clusterInfo)
             throws Exception {
-        List<String> clusters = getPulsarClusters(restTemplate, clusterInfo);
+        List<String> clusters = getClusters(restTemplate, clusterInfo);
         List<String> brokers = new ArrayList<>();
-        for (String brokerName : brokers) {
+        for (String brokerName : clusters) {
             String url = clusterInfo.getAdminUrl() + QUERY_BROKERS_PATH + "/" + brokerName;
-            clusters.addAll(
+            brokers.addAll(
                     HttpUtils.request(restTemplate, url, HttpMethod.GET, null, getHttpHeaders(clusterInfo.getToken()),
                             ArrayList.class));
         }
-        return clusters;
+        return brokers;
     }
 
     /**
@@ -135,7 +135,7 @@ public static List<String> getPulsarBrokers(RestTemplate restTemplate, PulsarClu
      * @return list of pulsar tenant infos
      * @throws Exception any exception if occurred
      */
-    public static List<String> getPulsarTenants(RestTemplate restTemplate, PulsarClusterInfo clusterInfo)
+    public static List<String> getTenants(RestTemplate restTemplate, PulsarClusterInfo clusterInfo)
             throws Exception {
         final String url = clusterInfo.getAdminUrl() + QUERY_TENANTS_PATH;
         return HttpUtils.request(restTemplate, url, HttpMethod.GET, null, getHttpHeaders(clusterInfo.getToken()),
@@ -151,7 +151,7 @@ public static List<String> getPulsarTenants(RestTemplate restTemplate, PulsarClu
      * @return list of pulsar namespace infos
      * @throws Exception any exception if occurred
      */
-    public static List<String> getPulsarNamespaces(RestTemplate restTemplate, PulsarClusterInfo clusterInfo,
+    public static List<String> getNamespaces(RestTemplate restTemplate, PulsarClusterInfo clusterInfo,
             String tenant) throws Exception {
         String url = clusterInfo.getAdminUrl() + QUERY_NAMESPACE_PATH + "/" + tenant;
         return HttpUtils.request(restTemplate, url, HttpMethod.GET, null, getHttpHeaders(clusterInfo.getToken()),
@@ -210,7 +210,7 @@ public static void createNamespace(RestTemplate restTemplate, PulsarClusterInfo
      * @return list of pulsar topic infos
      * @throws Exception any exception if occurred
      */
-    public static List<String> getPulsarTopics(RestTemplate restTemplate, PulsarClusterInfo clusterInfo, String tenant,
+    public static List<String> getTopics(RestTemplate restTemplate, PulsarClusterInfo clusterInfo, String tenant,
             String namespace) throws Exception {
         String url = clusterInfo.getAdminUrl() + QUERY_PERSISTENT_PATH + "/" + tenant + "/" + namespace;
         return HttpUtils.request(restTemplate, url, HttpMethod.GET, null, getHttpHeaders(clusterInfo.getToken()),
@@ -227,7 +227,7 @@ public static List<String> getPulsarTopics(RestTemplate restTemplate, PulsarClus
      * @return list of pulsar partitioned topic infos
      * @throws Exception any exception if occurred
      */
-    public static List<String> getPulsarPartitionedTopics(RestTemplate restTemplate, PulsarClusterInfo clusterInfo,
+    public static List<String> getPartitionedTopics(RestTemplate restTemplate, PulsarClusterInfo clusterInfo,
             String tenant, String namespace) throws Exception {
         String url =
                 clusterInfo.getAdminUrl() + QUERY_PERSISTENT_PATH + "/" + tenant + "/" + namespace + "/partitioned";
@@ -273,7 +273,7 @@ public static void createPartitionedTopic(RestTemplate restTemplate, PulsarClust
      * @return pulsar internal stat info of partitioned topic
      * @throws Exception any exception if occurred
      */
-    public static JsonObject getPulsarStatsPartitionedTopics(RestTemplate restTemplate,
+    public static JsonObject getStatsPartitionedTopics(RestTemplate restTemplate,
             PulsarClusterInfo clusterInfo, String topicPath) throws Exception {
         String url = clusterInfo.getAdminUrl() + QUERY_PERSISTENT_PATH + "/" + topicPath + "/partitioned-internalStats";
         return HttpUtils.request(restTemplate, url, HttpMethod.GET, null, getHttpHeaders(clusterInfo.getToken()),
@@ -289,7 +289,7 @@ public static JsonObject getPulsarStatsPartitionedTopics(RestTemplate restTempla
      * @return pulsar topic metadata info
      * @throws Exception any exception if occurred
      */
-    public static PulsarTopicMetadata getPulsarPartitionedTopicMetadata(RestTemplate restTemplate,
+    public static PulsarTopicMetadata getPartitionedTopicMetadata(RestTemplate restTemplate,
             PulsarClusterInfo clusterInfo, String topicPath) throws Exception {
         String url = clusterInfo.getAdminUrl() + QUERY_PERSISTENT_PATH + "/" + topicPath + "/partitions";
         return HttpUtils.request(restTemplate, url, HttpMethod.GET, null, getHttpHeaders(clusterInfo.getToken()),
@@ -421,7 +421,7 @@ public static String lookupTopic(RestTemplate restTemplate, PulsarClusterInfo cl
      */
     public static Map<String, String> lookupPartitionedTopic(RestTemplate restTemplate, PulsarClusterInfo clusterInfo,
             String topicPath) throws Exception {
-        PulsarTopicMetadata metadata = getPulsarPartitionedTopicMetadata(restTemplate, clusterInfo, topicPath);
+        PulsarTopicMetadata metadata = getPartitionedTopicMetadata(restTemplate, clusterInfo, topicPath);
         Map<String, String> map = new LinkedHashMap<>();
         for (int i = 0; i < metadata.getPartitions(); i++) {
             String partitionTopicName = topicPath + "-partition-" + i;