[#3702]refactor(API): Refactoring SupportsCatalogs.listCatalogs() method to return String[] (#3741)

### What changes were proposed in this pull request?

Currently, the SupportsCatalogs.listCatalogs() method returns an array
of NameIdentifier to represents the catalogs. Actually a String[] will
be more clear and easier to use.

### Why are the changes needed?

To make the API more clear to use.

Fix: #3702

### Does this PR introduce _any_ user-facing change?

Almost not; Just return type changed, will be more simple.

### How was this patch tested?

Yes, many test cases cover this API.

Author: shaofengshi

api/src/main/java/com/datastrato/gravitino/SupportsCatalogs.java

@@ -18,12 +18,12 @@
 public interface SupportsCatalogs {
 
   /**
-   * List all catalogs in the metalake.
+   * List the name of all catalogs in the metalake.
    *
-   * @return The list of catalog's name identifiers.
-   * @throws NoSuchMetalakeException If the metalake with namespace does not exist.
+   * @return The list of catalog's names.
+   * @throws NoSuchMetalakeException If the metalake does not exist.
    */
-  NameIdentifier[] listCatalogs() throws NoSuchMetalakeException;
+  String[] listCatalogs() throws NoSuchMetalakeException;
 
   /**
    * List all catalogs with their information in the metalake.

catalogs/catalog-hive/src/test/java/com/datastrato/gravitino/catalog/hive/integration/test/CatalogHiveIT.java

@@ -207,8 +207,8 @@ public static void stop() throws IOException {
             }));
     Arrays.stream(metalake.listCatalogs())
         .forEach(
-            (ident -> {
-              metalake.dropCatalog(ident.name());
+            (catalogName -> {
+              metalake.dropCatalog(catalogName);
             }));
     if (client != null) {
       client.dropMetalake(metalakeName);

catalogs/catalog-kafka/src/test/java/com/datastrato/gravitino/catalog/kafka/integration/test/CatalogKafkaIT.java

@@ -108,8 +108,8 @@ public static void shutdown() {
             }));
     Arrays.stream(metalake.listCatalogs())
         .forEach(
-            (ident -> {
-              metalake.dropCatalog(ident.name());
+            (catalogName -> {
+              metalake.dropCatalog(catalogName);
             }));
     client.dropMetalake(METALAKE_NAME);
     if (adminClient != null) {

clients/client-java/src/main/java/com/datastrato/gravitino/client/GravitinoClient.java

@@ -7,7 +7,6 @@
 
 import com.datastrato.gravitino.Catalog;
 import com.datastrato.gravitino.CatalogChange;
-import com.datastrato.gravitino.NameIdentifier;
 import com.datastrato.gravitino.SupportsCatalogs;
 import com.datastrato.gravitino.exceptions.CatalogAlreadyExistsException;
 import com.datastrato.gravitino.exceptions.NoSuchCatalogException;
@@ -58,7 +57,7 @@ private GravitinoMetalake getMetalake() {
   }
 
   @Override
-  public NameIdentifier[] listCatalogs() throws NoSuchMetalakeException {
+  public String[] listCatalogs() throws NoSuchMetalakeException {
     return getMetalake().listCatalogs();
   }
 

clients/client-java/src/main/java/com/datastrato/gravitino/client/GravitinoMetalake.java

@@ -52,11 +52,11 @@ public class GravitinoMetalake extends MetalakeDTO implements SupportsCatalogs {
   /**
    * List all the catalogs under this metalake.
    *
-   * @return A list of {@link NameIdentifier} of the catalogs under the specified namespace.
-   * @throws NoSuchMetalakeException if the metalake with specified namespace does not exist.
+   * @return A list of the catalog names under the current metalake.
+   * @throws NoSuchMetalakeException If the metalake does not exist.
    */
   @Override
-  public NameIdentifier[] listCatalogs() throws NoSuchMetalakeException {
+  public String[] listCatalogs() throws NoSuchMetalakeException {
 
     EntityListResponse resp =
         restClient.get(
@@ -66,7 +66,7 @@ public NameIdentifier[] listCatalogs() throws NoSuchMetalakeException {
             ErrorHandlers.catalogErrorHandler());
     resp.validate();
 
-    return resp.identifiers();
+    return Arrays.stream(resp.identifiers()).map(NameIdentifier::name).toArray(String[]::new);
   }
 
   /**

clients/client-java/src/test/java/com/datastrato/gravitino/client/TestGravitinoMetalake.java

@@ -76,16 +76,16 @@ public void testListCatalogs() throws JsonProcessingException {
 
     EntityListResponse resp = new EntityListResponse(new NameIdentifier[] {ident1, ident2});
     buildMockResource(Method.GET, path, null, resp, HttpStatus.SC_OK);
-    NameIdentifier[] catalogs = gravitinoClient.listCatalogs();
+    String[] catalogs = gravitinoClient.listCatalogs();
 
     Assertions.assertEquals(2, catalogs.length);
-    Assertions.assertEquals(ident1, catalogs[0]);
-    Assertions.assertEquals(ident2, catalogs[1]);
+    Assertions.assertEquals(ident1.name(), catalogs[0]);
+    Assertions.assertEquals(ident2.name(), catalogs[1]);
 
     // Test return empty catalog list
     EntityListResponse resp1 = new EntityListResponse(new NameIdentifier[] {});
     buildMockResource(Method.GET, path, null, resp1, HttpStatus.SC_OK);
-    NameIdentifier[] catalogs1 = gravitinoClient.listCatalogs();
+    String[] catalogs1 = gravitinoClient.listCatalogs();
     Assertions.assertEquals(0, catalogs1.length);
 
     // Test return internal error

flink-connector/src/main/java/com/datastrato/gravitino/flink/connector/catalog/GravitinoCatalogManager.java

@@ -5,14 +5,13 @@
 package com.datastrato.gravitino.flink.connector.catalog;
 
 import com.datastrato.gravitino.Catalog;
-import com.datastrato.gravitino.NameIdentifier;
 import com.datastrato.gravitino.client.GravitinoAdminClient;
 import com.datastrato.gravitino.client.GravitinoMetalake;
 import com.google.common.base.Preconditions;
+import com.google.common.collect.Sets;
 import java.util.Arrays;
 import java.util.Map;
 import java.util.Set;
-import java.util.stream.Collectors;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -128,12 +127,12 @@ public boolean dropCatalog(String catalogName) {
    * @return Set of catalog names
    */
   public Set<String> listCatalogs() {
-    NameIdentifier[] catalogNames = metalake.listCatalogs();
+    String[] catalogNames = metalake.listCatalogs();
     LOG.info(
         "Load metalake {}'s catalogs. catalogs: {}.",
         metalake.name(),
         Arrays.toString(catalogNames));
-    return Arrays.stream(catalogNames).map(NameIdentifier::name).collect(Collectors.toSet());
+    return Sets.newHashSet(catalogNames);
   }
 
   /**

integration-test/src/test/java/com/datastrato/gravitino/integration/test/trino/TrinoQueryIT.java

@@ -79,8 +79,8 @@ public static void setup() throws Exception {
   private static void cleanupTestEnv() throws Exception {
     try {
       Arrays.stream(TrinoQueryITBase.metalake.listCatalogs())
-          .filter(catalog -> catalog.name().startsWith("gt_"))
-          .forEach(catalog -> TrinoQueryITBase.dropCatalog(catalog.name()));
+          .filter(catalog -> catalog.startsWith("gt_"))
+          .forEach(TrinoQueryITBase::dropCatalog);
 
       await()
           .atMost(30, TimeUnit.SECONDS)

trino-connector/src/main/java/com/datastrato/gravitino/trino/connector/catalog/CatalogConnectorManager.java

@@ -142,7 +142,7 @@ private GravitinoMetalake retrieveMetalake(String metalakeName) {
   }
 
   private void loadCatalogs(GravitinoMetalake metalake) {
-    NameIdentifier[] catalogNames;
+    String[] catalogNames;
     try {
       catalogNames = metalake.listCatalogs();
     } catch (Exception e) {
@@ -159,10 +159,7 @@ private void loadCatalogs(GravitinoMetalake metalake) {
     Set<String> catalogNameStrings =
         Arrays.stream(catalogNames)
             .map(
-                id ->
-                    config.simplifyCatalogNames()
-                        ? id.name()
-                        : getTrinoCatalogName(metalake.name(), id.name()))
+                id -> config.simplifyCatalogNames() ? id : getTrinoCatalogName(metalake.name(), id))
             .collect(Collectors.toSet());
 
     for (Map.Entry<String, CatalogConnectorContext> entry : catalogConnectors.entrySet()) {
@@ -181,9 +178,9 @@ private void loadCatalogs(GravitinoMetalake metalake) {
     // Load new catalogs belows to the metalake.
     Arrays.stream(catalogNames)
         .forEach(
-            (NameIdentifier nameIdentifier) -> {
+            (String catalogName) -> {
               try {
-                Catalog catalog = metalake.loadCatalog(nameIdentifier.name());
+                Catalog catalog = metalake.loadCatalog(catalogName);
                 GravitinoCatalog gravitinoCatalog = new GravitinoCatalog(metalake.name(), catalog);
                 if (catalogConnectors.containsKey(getTrinoCatalogName(gravitinoCatalog))) {
                   // Reload catalogs that have been updated in Gravitino server.
@@ -195,7 +192,7 @@ private void loadCatalogs(GravitinoMetalake metalake) {
                 }
               } catch (Exception e) {
                 LOG.error(
-                    "Failed to load metalake {}'s catalog {}.", metalake.name(), nameIdentifier, e);
+                    "Failed to load metalake {}'s catalog {}.", metalake.name(), catalogName, e);
               }
             });
   }

trino-connector/src/test/java/com/datastrato/gravitino/trino/connector/GravitinoMockServer.java

@@ -129,12 +129,10 @@ private GravitinoMetalake createMetalake(String metalakeName) {
     when(metaLake.name()).thenReturn(metalakeName);
     when(metaLake.listCatalogs())
         .thenAnswer(
-            new Answer<NameIdentifier[]>() {
+            new Answer<String[]>() {
               @Override
-              public NameIdentifier[] answer(InvocationOnMock invocation) throws Throwable {
-                return metalakes.get(metalakeName).catalogs.keySet().stream()
-                    .map(catalogName -> NameIdentifier.of(metalakeName, catalogName))
-                    .toArray(NameIdentifier[]::new);
+              public String[] answer(InvocationOnMock invocation) throws Throwable {
+                return metalakes.get(metalakeName).catalogs.keySet().toArray(String[]::new);
               };
             });