Address comments

core/src/main/java/com/datastrato/gravitino/GravitinoEnv.java

@@ -110,17 +110,6 @@ public void setLockManager(LockManager lockManager) {
     this.lockManager = lockManager;
   }
 
-  /**
-   * This method is used for testing purposes only to set the access manager for test in package
-   * `com.datastrato.gravitino.server.web.rest` and `com.datastrato.gravitino.authorization`.
-   *
-   * @param accessControlManager The access control manager to be set.
-   */
-  @VisibleForTesting
-  public void setAccessControlManager(AccessControlManager accessControlManager) {
-    this.accessControlManager = accessControlManager;
-  }
-
   /**
    * Initialize the Gravitino environment.
    *

core/src/main/java/com/datastrato/gravitino/catalog/EntityCombinedSchema.java

@@ -6,9 +6,9 @@
 
 import com.datastrato.gravitino.Audit;
 import com.datastrato.gravitino.Schema;
+import com.datastrato.gravitino.StringIdentifier;
 import com.datastrato.gravitino.meta.AuditInfo;
 import com.datastrato.gravitino.meta.SchemaEntity;
-import java.util.Collections;
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -24,6 +24,8 @@ public final class EntityCombinedSchema implements Schema {
 
   // Sets of properties that should be hidden from the user.
   private Set<String> hiddenProperties;
+  // If imported is true, it means that storage backend have stored the correct entity.
+  // Otherwise, we should import the external entity to the storage backend.
   private boolean imported;
 
   private EntityCombinedSchema(Schema schema, SchemaEntity schemaEntity) {
@@ -85,7 +87,7 @@ public boolean imported() {
     return imported;
   }
 
-  Map<String, String> schemaProperties() {
-    return Collections.unmodifiableMap(schema.properties());
+  StringIdentifier stringIdentifier() {
+    return StringIdentifier.fromProperties(schema.properties());
   }
 }

core/src/main/java/com/datastrato/gravitino/catalog/EntityCombinedTable.java

@@ -5,6 +5,7 @@
 package com.datastrato.gravitino.catalog;
 
 import com.datastrato.gravitino.Audit;
+import com.datastrato.gravitino.StringIdentifier;
 import com.datastrato.gravitino.meta.AuditInfo;
 import com.datastrato.gravitino.meta.TableEntity;
 import com.datastrato.gravitino.rel.Column;
@@ -14,7 +15,6 @@
 import com.datastrato.gravitino.rel.expressions.sorts.SortOrder;
 import com.datastrato.gravitino.rel.expressions.transforms.Transform;
 import com.datastrato.gravitino.rel.indexes.Index;
-import java.util.Collections;
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -31,6 +31,8 @@ public final class EntityCombinedTable implements Table {
 
   // Sets of properties that should be hidden from the user.
   private Set<String> hiddenProperties;
+  // If imported is true, it means that storage backend have stored the correct entity.
+  // Otherwise, we should import the external entity to the storage backend.
   private boolean imported;
 
   private EntityCombinedTable(Table table, TableEntity tableEntity) {
@@ -122,7 +124,7 @@ public Audit auditInfo() {
         : mergedAudit.merge(tableEntity.auditInfo(), true /* overwrite */);
   }
 
-  Map<String, String> tableProperties() {
-    return Collections.unmodifiableMap(table.properties());
+  StringIdentifier stringIdentifier() {
+    return StringIdentifier.fromProperties(table.properties());
   }
 }

core/src/main/java/com/datastrato/gravitino/catalog/EntityCombinedTopic.java

@@ -5,10 +5,10 @@
 package com.datastrato.gravitino.catalog;
 
 import com.datastrato.gravitino.Audit;
+import com.datastrato.gravitino.StringIdentifier;
 import com.datastrato.gravitino.messaging.Topic;
 import com.datastrato.gravitino.meta.AuditInfo;
 import com.datastrato.gravitino.meta.TopicEntity;
-import java.util.Collections;
 import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
@@ -24,6 +24,8 @@ public class EntityCombinedTopic implements Topic {
 
   // Sets of properties that should be hidden from the user.
   private Set<String> hiddenProperties;
+  // If imported is true, it means that storage backend have stored the correct entity.
+  // Otherwise, we should import the external entity to the storage backend.
   private boolean imported;
 
   private EntityCombinedTopic(Topic topic, TopicEntity topicEntity) {
@@ -85,7 +87,7 @@ public boolean imported() {
     return imported;
   }
 
-  Map<String, String> topicProperties() {
-    return Collections.unmodifiableMap(topic.properties());
+  StringIdentifier stringIdentifier() {
+    return StringIdentifier.fromProperties(topic.properties());
   }
 }

core/src/main/java/com/datastrato/gravitino/catalog/OperationDispatcher.java

@@ -6,6 +6,7 @@
 
 import static com.datastrato.gravitino.catalog.PropertiesMetadataHelpers.validatePropertyForAlter;
 
+import com.datastrato.gravitino.Entity;
 import com.datastrato.gravitino.EntityStore;
 import com.datastrato.gravitino.HasIdentifier;
 import com.datastrato.gravitino.NameIdentifier;
@@ -272,6 +273,15 @@ boolean isManagedEntity(NameIdentifier catalogIdent, Capability.Scope scope) {
         IllegalArgumentException.class);
   }
 
+  boolean isEntityExist(NameIdentifier ident, Entity.EntityType type) {
+    try {
+      return store.exists(ident, type);
+    } catch (Exception e) {
+      LOG.error(FormattedErrorMessages.STORE_OP_FAILURE, "exists", ident, e);
+      throw new RuntimeException("Fail to access underlying storage", e);
+    }
+  }
+
   static final class FormattedErrorMessages {
     static final String STORE_OP_FAILURE =
         "Failed to {} entity for {} in "

core/src/main/java/com/datastrato/gravitino/catalog/SchemaOperationDispatcher.java

@@ -310,14 +310,20 @@ private boolean importSchema(NameIdentifier identifier) {
       return false;
     }
 
-    StringIdentifier stringId = getStringIdFromProperties(combinedSchema.schemaProperties());
+    StringIdentifier stringId = null;
+    try {
+      stringId = combinedSchema.stringIdentifier();
+    } catch (IllegalArgumentException ie) {
+      LOG.warn(FormattedErrorMessages.STRING_ID_PARSE_ERROR, ie.getMessage());
+    }
+
     long uid;
     if (stringId != null) {
-      // If the entity in the store doesn't match the external system, we use the data
+      // If the entity in the store doesn't match the one in the external system, we use the data
       // of external system to correct it.
       uid = stringId.id();
     } else {
-      // If store doesn't exist entity, we sync the entity from the external system.
+      // If entity doesn't exist, we import the entity from the external system.
       uid = idGenerator.nextId();
     }
 
@@ -361,6 +367,8 @@ private EntityCombinedSchema loadCombinedSchema(NameIdentifier ident) {
                   catalogIdentifier,
                   HasPropertyMetadata::schemaPropertiesMetadata,
                   schema.properties()))
+          // The meta of managed schema is stored by Gravitino,
+          // We don't need to import it again.
           .withImported(true);
     }
 
@@ -373,7 +381,7 @@ private EntityCombinedSchema loadCombinedSchema(NameIdentifier ident) {
                   catalogIdentifier,
                   HasPropertyMetadata::schemaPropertiesMetadata,
                   schema.properties()))
-          .withImported(isEntityExist(ident));
+          .withImported(isEntityExist(ident, SCHEMA));
     }
 
     SchemaEntity schemaEntity =
@@ -393,13 +401,4 @@ private EntityCombinedSchema loadCombinedSchema(NameIdentifier ident) {
                 schema.properties()))
         .withImported(imported);
   }
-
-  private boolean isEntityExist(NameIdentifier ident) {
-    try {
-      return store.exists(ident, SCHEMA);
-    } catch (Exception e) {
-      LOG.error(FormattedErrorMessages.STORE_OP_FAILURE, "exists", ident, e);
-      throw new RuntimeException("Fail to access underlying storage");
-    }
-  }
 }

core/src/main/java/com/datastrato/gravitino/catalog/TableOperationDispatcher.java

@@ -374,14 +374,20 @@ private boolean importTable(NameIdentifier identifier) {
       return false;
     }
 
-    StringIdentifier stringId = getStringIdFromProperties(combinedTable.tableProperties());
+    StringIdentifier stringId = null;
+    try {
+      stringId = combinedTable.stringIdentifier();
+    } catch (IllegalArgumentException ie) {
+      LOG.warn(FormattedErrorMessages.STRING_ID_PARSE_ERROR, ie.getMessage());
+    }
+
     long uid;
     if (stringId != null) {
       // If the entity in the store doesn't match the external system, we use the data
       // of external system to correct it.
       uid = stringId.id();
     } else {
-      // If store doesn't exist entity, we sync the entity from the external system.
+      // If entity doesn't exist, we import the entity from the external system.
       uid = idGenerator.nextId();
     }
 
@@ -407,15 +413,6 @@ private boolean importTable(NameIdentifier identifier) {
     return true;
   }
 
-  private boolean isEntityExist(NameIdentifier ident) {
-    try {
-      return store.exists(ident, TABLE);
-    } catch (Exception e) {
-      LOG.error(FormattedErrorMessages.STORE_OP_FAILURE, "exists", ident, e);
-      throw new RuntimeException("Fail to access underlying storage");
-    }
-  }
-
   private EntityCombinedTable loadCombinedTable(NameIdentifier ident) {
     NameIdentifier catalogIdentifier = getCatalogIdentifier(ident);
     Table table =
@@ -425,15 +422,19 @@ private EntityCombinedTable loadCombinedTable(NameIdentifier ident) {
             NoSuchTableException.class);
 
     StringIdentifier stringId = getStringIdFromProperties(table.properties());
-    // Case 1: The table is not created by Gravitino.
+    // Case 1: The table is not created by Gravitino or the backend storage does not support storing
+    // string identifier.
     if (stringId == null) {
       return EntityCombinedTable.of(table)
           .withHiddenPropertiesSet(
               getHiddenPropertyNames(
                   catalogIdentifier,
                   HasPropertyMetadata::tablePropertiesMetadata,
                   table.properties()))
-          .withImported(isEntityExist(ident));
+          // Some tables don't have properties or are not created by Gravitino,
+          // we can't use stringIdentifier to judge whether schema is ever imported or not.
+          // We need to check whether the entity exists.
+          .withImported(isEntityExist(ident, TABLE));
     }
 
     TableEntity tableEntity =
@@ -443,6 +444,8 @@ private EntityCombinedTable loadCombinedTable(NameIdentifier ident) {
             "GET",
             stringId.id());
 
+    // If the entity is inconsistent from the one of the external system,
+    // we should import it.
     boolean imported = tableEntity != null;
 
     return EntityCombinedTable.of(table, tableEntity)

core/src/main/java/com/datastrato/gravitino/catalog/TopicOperationDispatcher.java

@@ -278,15 +278,20 @@ private boolean importTopic(NameIdentifier identifier) {
       return false;
     }
 
-    StringIdentifier stringId = getStringIdFromProperties(combinedTopic.topicProperties());
+    StringIdentifier stringId = null;
+    try {
+      stringId = combinedTopic.stringIdentifier();
+    } catch (IllegalArgumentException ie) {
+      LOG.warn(FormattedErrorMessages.STRING_ID_PARSE_ERROR, ie.getMessage());
+    }
 
     long uid;
     if (stringId != null) {
       // If the entity in the store doesn't match the external system, we use the data
       // of external system to correct it.
       uid = stringId.id();
     } else {
-      // If store doesn't exist entity, we sync the entity from the external system.
+      // If entity doesn't exist, we import the entity from the external system.
       uid = idGenerator.nextId();
     }
 
@@ -315,15 +320,6 @@ private boolean importTopic(NameIdentifier identifier) {
     return true;
   }
 
-  private boolean isEntityExist(NameIdentifier ident) {
-    try {
-      return store.exists(ident, TOPIC);
-    } catch (Exception e) {
-      LOG.error(FormattedErrorMessages.STORE_OP_FAILURE, "exists", ident, e);
-      throw new RuntimeException("Fail to access underlying storage");
-    }
-  }
-
   private EntityCombinedTopic loadCombinedTopic(NameIdentifier ident) {
     NameIdentifier catalogIdent = getCatalogIdentifier(ident);
     Topic topic =
@@ -342,7 +338,7 @@ private EntityCombinedTopic loadCombinedTopic(NameIdentifier ident) {
           .withHiddenPropertiesSet(
               getHiddenPropertyNames(
                   catalogIdent, HasPropertyMetadata::topicPropertiesMetadata, topic.properties()))
-          .withImported(isEntityExist(ident));
+          .withImported(isEntityExist(ident, TOPIC));
     }
 
     TopicEntity topicEntity =

core/src/test/java/com/datastrato/gravitino/catalog/TestSchemaNormalizeDispatcher.java

@@ -21,7 +21,7 @@ public class TestSchemaNormalizeDispatcher extends TestOperationDispatcher {
   private static SchemaNormalizeDispatcher schemaNormalizeDispatcher;
 
   @BeforeAll
-  public static void initialize() throws IOException {
+  public static void initialize() throws IOException, IllegalAccessException {
     TestSchemaOperationDispatcher.initialize();
     schemaNormalizeDispatcher =
         new SchemaNormalizeDispatcher(TestSchemaOperationDispatcher.dispatcher);

core/src/test/java/com/datastrato/gravitino/catalog/TestSchemaOperationDispatcher.java

@@ -34,6 +34,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
+import org.apache.commons.lang3.reflect.FieldUtils;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeAll;
 import org.junit.jupiter.api.Test;
@@ -43,14 +44,14 @@ public class TestSchemaOperationDispatcher extends TestOperationDispatcher {
   static SchemaOperationDispatcher dispatcher;
 
   @BeforeAll
-  public static void initialize() throws IOException {
+  public static void initialize() throws IOException, IllegalAccessException {
     dispatcher = new SchemaOperationDispatcher(catalogManager, entityStore, idGenerator);
 
     Config config = mock(Config.class);
     doReturn(100000L).when(config).get(Configs.TREE_LOCK_MAX_NODE_IN_MEMORY);
     doReturn(1000L).when(config).get(Configs.TREE_LOCK_MIN_NODE_IN_MEMORY);
     doReturn(36000L).when(config).get(Configs.TREE_LOCK_CLEAN_INTERVAL);
-    GravitinoEnv.getInstance().setLockManager(new LockManager(config));
+    FieldUtils.writeField(GravitinoEnv.getInstance(), "lockManager", new LockManager(config), true);
   }
 
   @Test

server/src/test/java/com/datastrato/gravitino/server/web/rest/TestCatalogOperations.java

@@ -46,6 +46,7 @@
 import javax.ws.rs.core.Application;
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
+import org.apache.commons.lang3.reflect.FieldUtils;
 import org.glassfish.hk2.utilities.binding.AbstractBinder;
 import org.glassfish.jersey.server.ResourceConfig;
 import org.glassfish.jersey.test.JerseyTest;
@@ -69,12 +70,12 @@ public HttpServletRequest get() {
   private CatalogManager manager = mock(CatalogManager.class);
 
   @BeforeAll
-  public static void setup() {
+  public static void setup() throws IllegalAccessException {
     Config config = mock(Config.class);
     Mockito.doReturn(100000L).when(config).get(TREE_LOCK_MAX_NODE_IN_MEMORY);
     Mockito.doReturn(1000L).when(config).get(TREE_LOCK_MIN_NODE_IN_MEMORY);
     Mockito.doReturn(36000L).when(config).get(TREE_LOCK_CLEAN_INTERVAL);
-    GravitinoEnv.getInstance().setLockManager(new LockManager(config));
+    FieldUtils.writeField(GravitinoEnv.getInstance(), "lockManager", new LockManager(config), true);
   }
 
   @Override

server/src/test/java/com/datastrato/gravitino/server/web/rest/TestFilesetOperations.java

@@ -44,6 +44,7 @@
 import javax.ws.rs.core.Application;
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
+import org.apache.commons.lang3.reflect.FieldUtils;
 import org.glassfish.jersey.internal.inject.AbstractBinder;
 import org.glassfish.jersey.server.ResourceConfig;
 import org.glassfish.jersey.test.JerseyTest;
@@ -72,12 +73,12 @@ public HttpServletRequest get() {
   private final String schema = "schema1";
 
   @BeforeAll
-  public static void setup() {
+  public static void setup() throws IllegalAccessException {
     Config config = mock(Config.class);
     Mockito.doReturn(100000L).when(config).get(TREE_LOCK_MAX_NODE_IN_MEMORY);
     Mockito.doReturn(1000L).when(config).get(TREE_LOCK_MIN_NODE_IN_MEMORY);
     Mockito.doReturn(36000L).when(config).get(TREE_LOCK_CLEAN_INTERVAL);
-    GravitinoEnv.getInstance().setLockManager(new LockManager(config));
+    FieldUtils.writeField(GravitinoEnv.getInstance(), "lockManager", new LockManager(config), true);
   }
 
   @Override