[#4242][#2267][#3091] improvement(catalogs): Make some catalog proper…

…ty from immutable to mutable. (#4262)

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

Change the properties of catalog from immutable to mutable. 

### Why are the changes needed?

This is to prevent users from assigning the property an incorrect value
by mistake.

Fix: #4242
Fix: #2267
Fix: #3091 

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

N/A.
### How was this patch tested?


UT & IT.

catalogs/catalog-hadoop/src/main/java/org/apache/gravitino/catalog/hadoop/HadoopCatalogPropertiesMetadata.java

@@ -41,7 +41,7 @@ public class HadoopCatalogPropertiesMetadata extends BaseCatalogPropertiesMetada
               PropertyEntry.stringOptionalPropertyEntry(
                   LOCATION,
                   "The storage location managed by Hadoop fileset catalog",
-                  true /* immutable */,
+                  false /* immutable */,
                   null,
                   false /* hidden */))
           .putAll(BASIC_CATALOG_PROPERTY_ENTRIES)

catalogs/catalog-hadoop/src/main/java/org/apache/gravitino/catalog/hadoop/authentication/AuthenticationConfig.java

@@ -88,19 +88,18 @@ public boolean isKerberosAuth() {
               PropertyEntry.booleanPropertyEntry(
                   IMPERSONATION_ENABLE_KEY,
                   "Whether to enable impersonation for the Hadoop catalog",
-                  false,
-                  true,
-                  KERBEROS_DEFAULT_IMPERSONATION_ENABLE,
-                  false,
-                  false))
+                  false /* required */,
+                  true /* immutable */,
+                  KERBEROS_DEFAULT_IMPERSONATION_ENABLE /* default value */,
+                  false /* hidden */,
+                  false /* reserved */))
           .put(
               AUTH_TYPE_KEY,
-              PropertyEntry.stringImmutablePropertyEntry(
+              PropertyEntry.stringOptionalPropertyEntry(
                   AUTH_TYPE_KEY,
                   "The type of authentication for Hadoop catalog, currently we only support simple and Kerberos",
-                  false,
-                  null,
-                  false,
-                  false))
+                  false /* immutable */,
+                  null /* default value */,
+                  false /* hidden */))
           .build();
 }

catalogs/catalog-hadoop/src/main/java/org/apache/gravitino/catalog/hadoop/authentication/kerberos/KerberosConfig.java

@@ -95,23 +95,35 @@ public int getFetchTimeoutSec() {
       new ImmutableMap.Builder<String, PropertyEntry<?>>()
           .put(
               KEY_TAB_URI_KEY,
-              PropertyEntry.stringImmutablePropertyEntry(
-                  KEY_TAB_URI_KEY, "The uri of key tab for the catalog", false, null, false, false))
+              PropertyEntry.stringOptionalPropertyEntry(
+                  KEY_TAB_URI_KEY,
+                  "The uri of key tab for the catalog",
+                  false /* immutable */,
+                  null /* default value */,
+                  false /* hidden */))
           .put(
               PRINCIPAL_KEY,
-              PropertyEntry.stringImmutablePropertyEntry(
-                  PRINCIPAL_KEY, "The principal for the catalog", false, null, false, false))
+              PropertyEntry.stringOptionalPropertyEntry(
+                  PRINCIPAL_KEY,
+                  "The principal for the catalog",
+                  false /* immutable */,
+                  null /* defaultValue */,
+                  false /* hidden */))
           .put(
               CHECK_INTERVAL_SEC_KEY,
               PropertyEntry.integerOptionalPropertyEntry(
                   CHECK_INTERVAL_SEC_KEY,
                   "The interval to check validness of the principal",
-                  true,
-                  60,
-                  false))
+                  true /* immutable */,
+                  60 /* defaultValue */,
+                  false /* hidden */))
           .put(
               FETCH_TIMEOUT_SEC_KEY,
               PropertyEntry.integerOptionalPropertyEntry(
-                  FETCH_TIMEOUT_SEC_KEY, "The timeout to fetch key tab", true, 60, false))
+                  FETCH_TIMEOUT_SEC_KEY,
+                  "The timeout to fetch key tab",
+                  false /* immutable */,
+                  60 /* defaultValue */,
+                  false /* hidden */))
           .build();
 }

catalogs/catalog-hadoop/src/test/java/org/apache/gravitino/catalog/hadoop/integration/test/HadoopCatalogIT.java

@@ -26,6 +26,7 @@
 import java.util.Comparator;
 import java.util.Map;
 import org.apache.gravitino.Catalog;
+import org.apache.gravitino.CatalogChange;
 import org.apache.gravitino.NameIdentifier;
 import org.apache.gravitino.Namespace;
 import org.apache.gravitino.Schema;
@@ -137,6 +138,29 @@ private static void dropSchema() {
     Assertions.assertFalse(catalog.asSchemas().schemaExists(schemaName));
   }
 
+  @Test
+  void testAlterCatalogLocation() {
+    String catalogName = GravitinoITUtils.genRandomName("test_alter_catalog_location");
+    String location = defaultBaseLocation();
+    String newLocation = location + "/new_location";
+
+    Map<String, String> catalogProperties = ImmutableMap.of("location", location);
+    // Create a catalog using location
+    Catalog filesetCatalog =
+        metalake.createCatalog(
+            catalogName, Catalog.Type.FILESET, provider, "comment", catalogProperties);
+
+    Assertions.assertEquals(location, filesetCatalog.properties().get("location"));
+
+    // Now try to alter the location and change it to `newLocation`.
+    Catalog modifiedCatalog =
+        metalake.alterCatalog(catalogName, CatalogChange.setProperty("location", newLocation));
+
+    Assertions.assertEquals(newLocation, modifiedCatalog.properties().get("location"));
+
+    metalake.dropCatalog(catalogName);
+  }
+
   @Test
   public void testCreateFileset() throws IOException {
     // create fileset

catalogs/catalog-hadoop/src/test/java/org/apache/gravitino/catalog/hadoop/integration/test/HadoopUserAuthenticationIT.java

@@ -36,6 +36,7 @@
 import java.util.Map;
 import org.apache.commons.io.FileUtils;
 import org.apache.gravitino.Catalog;
+import org.apache.gravitino.CatalogChange;
 import org.apache.gravitino.Configs;
 import org.apache.gravitino.NameIdentifier;
 import org.apache.gravitino.SchemaChange;
@@ -271,6 +272,20 @@ public void testUserAuthentication() {
     Assertions.assertTrue(
         exceptionMessage.contains("Permission denied: user=gravitino_client, access=WRITE"));
 
+    // Make the property wrong by changing the principal
+    gravitinoMetalake.alterCatalog(
+        CATALOG_NAME, CatalogChange.setProperty(PRINCIPAL_KEY, HADOOP_CLIENT_PRINCIPAL + "wrong"));
+    exception =
+        Assertions.assertThrows(
+            Exception.class,
+            () -> catalog.asSchemas().createSchema(SCHEMA_NAME, "comment", ImmutableMap.of()));
+    exceptionMessage = Throwables.getStackTraceAsString(exception);
+    Assertions.assertTrue(exceptionMessage.contains("Failed to login with Kerberos"));
+
+    // Restore the property, everything goes okay.
+    gravitinoMetalake.alterCatalog(
+        CATALOG_NAME, CatalogChange.setProperty(PRINCIPAL_KEY, HADOOP_CLIENT_PRINCIPAL));
+
     // Now try to give the user the permission to create schema again
     kerberosHiveContainer.executeInContainer("hadoop", "fs", "-chmod", "-R", "777", "/user/hadoop");
     Assertions.assertDoesNotThrow(

catalogs/catalog-hive/src/main/java/org/apache/gravitino/catalog/hive/HiveCatalogPropertiesMeta.java

@@ -59,63 +59,74 @@ public class HiveCatalogPropertiesMeta extends BaseCatalogPropertiesMetadata {
           .put(
               METASTORE_URIS,
               PropertyEntry.stringRequiredPropertyEntry(
-                  METASTORE_URIS, "The Hive metastore URIs", true, false))
+                  METASTORE_URIS,
+                  "The Hive metastore URIs",
+                  false /* immutable */,
+                  false /* hidden */))
           .put(
               CLIENT_POOL_SIZE,
               PropertyEntry.integerOptionalPropertyEntry(
                   CLIENT_POOL_SIZE,
                   "The maximum number of Hive metastore clients in the pool for Gravitino",
-                  true,
+                  false /* immutable */,
                   DEFAULT_CLIENT_POOL_SIZE,
-                  false))
+                  false /* hidden */))
           .put(
               CLIENT_POOL_CACHE_EVICTION_INTERVAL_MS,
               PropertyEntry.longOptionalPropertyEntry(
                   CLIENT_POOL_CACHE_EVICTION_INTERVAL_MS,
                   "The cache pool eviction interval",
-                  true,
+                  false /* immutable */,
                   DEFAULT_CLIENT_POOL_CACHE_EVICTION_INTERVAL_MS,
-                  false))
+                  false /* hidden */))
           .put(
               IMPERSONATION_ENABLE,
               PropertyEntry.booleanPropertyEntry(
                   IMPERSONATION_ENABLE,
                   "Enable user impersonation for Hive catalog",
-                  false,
-                  true,
+                  false /* Whether this property is required */,
+                  false /* immutable */,
                   DEFAULT_IMPERSONATION_ENABLE,
-                  false,
-                  false))
+                  false /* hidden */,
+                  false /* reserved */))
           .put(
               KEY_TAB_URI,
-              PropertyEntry.stringImmutablePropertyEntry(
-                  KEY_TAB_URI, "The uri of key tab for the catalog", false, null, false, false))
+              PropertyEntry.stringOptionalPropertyEntry(
+                  KEY_TAB_URI,
+                  "The uri of key tab for the catalog",
+                  false /* immutable */,
+                  null /* defaultValue */,
+                  false /* hidden */))
           .put(
               PRINCIPAL,
-              PropertyEntry.stringImmutablePropertyEntry(
-                  PRINCIPAL, "The principal for the catalog", false, null, false, false))
+              PropertyEntry.stringOptionalPropertyEntry(
+                  PRINCIPAL,
+                  "The principal for the catalog",
+                  false /* immutable */,
+                  null /* defaultValue */,
+                  false /* hidden */))
           .put(
               CHECK_INTERVAL_SEC,
               PropertyEntry.integerOptionalPropertyEntry(
                   CHECK_INTERVAL_SEC,
                   "The interval to check validness of the principal",
-                  true,
-                  60,
-                  false))
+                  false /* immutable */,
+                  60 /* defaultValue */,
+                  false /* hidden */))
           .put(
               FETCH_TIMEOUT_SEC,
               PropertyEntry.integerOptionalPropertyEntry(
-                  FETCH_TIMEOUT_SEC, "The timeout to fetch key tab", true, 60, false))
+                  FETCH_TIMEOUT_SEC, "The timeout to fetch key tab", false, 60, false))
           .put(
               LIST_ALL_TABLES,
               PropertyEntry.booleanPropertyEntry(
                   LIST_ALL_TABLES,
                   "Lists all tables in a database, including non-Hive tables, such as Iceberg, etc.",
-                  false,
-                  false,
+                  false /* required */,
+                  false /* immutable */,
                   DEFAULT_LIST_ALL_TABLES,
-                  false,
-                  false))
+                  false /* hidden */,
+                  false /* reserved */))
           .putAll(BASIC_CATALOG_PROPERTY_ENTRIES)
           .build();
 

catalogs/catalog-hive/src/test/java/org/apache/gravitino/catalog/hive/integration/test/CatalogHiveIT.java

@@ -1637,6 +1637,40 @@ void testCustomCatalogOperations() {
                 catalogName + "_not_exists", "org.apache.gravitino.catalog.not.exists"));
   }
 
+  @Test
+  void testAlterCatalogProperties() {
+    Map<String, String> properties = Maps.newHashMap();
+    String nameOfCatalog = GravitinoITUtils.genRandomName("catalog");
+    // Wrong Hive HIVE_METASTORE_URIS
+    String wrongHiveMetastoreURI = HIVE_METASTORE_URIS + "_wrong";
+    properties.put(METASTORE_URIS, wrongHiveMetastoreURI);
+    Catalog createdCatalog =
+        metalake.createCatalog(
+            nameOfCatalog, Catalog.Type.RELATIONAL, provider, "comment", properties);
+    Assertions.assertEquals(wrongHiveMetastoreURI, createdCatalog.properties().get(METASTORE_URIS));
+
+    // As it's wrong metastore uri, it should throw exception.
+    Exception exception =
+        Assertions.assertThrows(
+            Exception.class,
+            () -> createdCatalog.asSchemas().createSchema("schema", "comment", ImmutableMap.of()));
+    Assertions.assertTrue(exception.getMessage().contains("Failed to connect to Hive Metastore"));
+
+    Catalog newCatalog =
+        metalake.alterCatalog(
+            nameOfCatalog, CatalogChange.setProperty(METASTORE_URIS, HIVE_METASTORE_URIS));
+    Assertions.assertEquals(HIVE_METASTORE_URIS, newCatalog.properties().get(METASTORE_URIS));
+
+    // The URI has restored, so it should not throw exception.
+    Assertions.assertDoesNotThrow(
+        () -> {
+          newCatalog.asSchemas().createSchema("schema", "comment", ImmutableMap.of());
+        });
+
+    newCatalog.asSchemas().dropSchema("schema", true);
+    metalake.dropCatalog(nameOfCatalog);
+  }
+
   private static void createCatalogWithCustomOperation(String catalogName, String customImpl) {
     Map<String, String> properties = Maps.newHashMap();
     properties.put(METASTORE_URIS, HIVE_METASTORE_URIS);

catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalogPropertiesMetadata.java

@@ -19,7 +19,8 @@
 package org.apache.gravitino.catalog.jdbc;
 
 import static org.apache.gravitino.connector.PropertyEntry.integerPropertyEntry;
-import static org.apache.gravitino.connector.PropertyEntry.stringImmutablePropertyEntry;
+import static org.apache.gravitino.connector.PropertyEntry.stringOptionalPropertyEntry;
+import static org.apache.gravitino.connector.PropertyEntry.stringPropertyEntry;
 
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Maps;
@@ -45,57 +46,54 @@ public class JdbcCatalogPropertiesMetadata extends BaseCatalogPropertiesMetadata
   static {
     List<PropertyEntry<?>> propertyEntries =
         ImmutableList.of(
-            stringImmutablePropertyEntry(
+            stringPropertyEntry(
                 JdbcConfig.JDBC_URL.getKey(),
                 JdbcConfig.JDBC_URL.getDoc(),
-                true,
-                null,
-                false,
-                false),
-            stringImmutablePropertyEntry(
+                true /* required */,
+                false /* immutable */,
+                null /* defaultValue */,
+                false /* hidden */,
+                false /* reserved */),
+            stringOptionalPropertyEntry(
                 JdbcConfig.JDBC_DATABASE.getKey(),
                 JdbcConfig.JDBC_DATABASE.getDoc(),
-                false,
-                null,
-                false,
-                false),
-            stringImmutablePropertyEntry(
+                false /* immutable */,
+                null /* defaultValue */,
+                false /* hidden */),
+            stringOptionalPropertyEntry(
                 JdbcConfig.JDBC_DRIVER.getKey(),
                 JdbcConfig.JDBC_DRIVER.getDoc(),
-                false,
-                null,
-                false,
-                false),
-            stringImmutablePropertyEntry(
+                false /* immutable */,
+                null /* defaultValue */,
+                false /* hidden */),
+            stringOptionalPropertyEntry(
                 JdbcConfig.USERNAME.getKey(),
                 JdbcConfig.USERNAME.getDoc(),
-                false,
-                null,
-                false,
-                false),
-            stringImmutablePropertyEntry(
+                false /* immutable */,
+                null /* defaultValue */,
+                false /* hidden */),
+            stringOptionalPropertyEntry(
                 JdbcConfig.PASSWORD.getKey(),
                 JdbcConfig.PASSWORD.getDoc(),
-                false,
-                null,
-                false,
-                false),
+                false /* immutable */,
+                null /* defaultValue */,
+                false /* hidden */),
             integerPropertyEntry(
                 JdbcConfig.POOL_MIN_SIZE.getKey(),
                 JdbcConfig.POOL_MIN_SIZE.getDoc(),
-                false,
-                true,
+                false /* required */,
+                false /* immutable */,
                 JdbcConfig.POOL_MIN_SIZE.getDefaultValue(),
-                true,
-                false),
+                true /* hidden */,
+                false /* reserved */),
             integerPropertyEntry(
                 JdbcConfig.POOL_MAX_SIZE.getKey(),
                 JdbcConfig.POOL_MAX_SIZE.getDoc(),
-                false,
-                true,
+                false /* required */,
+                false /* immutable */,
                 JdbcConfig.POOL_MAX_SIZE.getDefaultValue(),
-                true,
-                false));
+                true /* hidden */,
+                false /* reserved */));
     PROPERTIES_METADATA = Maps.uniqueIndex(propertyEntries, PropertyEntry::getName);
   }