[Fix](Catalog) Close system resources when dropping catalog

fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogIf.java

@@ -94,7 +94,7 @@ default String getResource() {
 
     default void notifyPropertiesUpdated(Map<String, String> updatedProps) {
         if (this instanceof ExternalCatalog) {
-            ((ExternalCatalog) this).onRefresh(false);
+            ((ExternalCatalog) this).resetToUninitialized(false);
         }
     }
 

fe/fe-core/src/main/java/org/apache/doris/datasource/CatalogMgr.java

@@ -125,7 +125,7 @@ private void addCatalog(CatalogIf catalog) {
         idToCatalog.put(catalog.getId(), catalog);
         String catalogName = catalog.getName();
         if (!catalogName.equals(InternalCatalog.INTERNAL_CATALOG_NAME)) {
-            ((ExternalCatalog) catalog).onRefresh(false);
+            ((ExternalCatalog) catalog).resetToUninitialized(false);
         }
         if (!Strings.isNullOrEmpty(catalog.getResource())) {
             Resource resource = Env.getCurrentEnv().getResourceMgr().getResource(catalog.getResource());

fe/fe-core/src/main/java/org/apache/doris/datasource/ExternalCatalog.java

@@ -501,7 +501,23 @@ private List<Pair<String, String>> getFilteredDatabaseNames() {
         return remoteToLocalPairs;
     }
 
-    public void onRefresh(boolean invalidCache) {
+    /**
+     * Resets the Catalog state to uninitialized, releases resources held by {@code initLocalObjectsImpl()}
+     * <p>
+     * This method is typically invoked during operations such as {@code CREATE CATALOG}
+     * and {@code MODIFY CATALOG}. It marks the object as uninitialized, clears cached
+     * configurations, and ensures that resources allocated during {@link #initLocalObjectsImpl()}
+     * are properly released via {@link #onClose()}
+     * </p>
+     * <p>
+     * The {@code onClose()} method is responsible for cleaning up resources that were initialized
+     * in {@code initLocalObjectsImpl()}, preventing potential resource leaks.
+     * </p>
+     *
+     * @param invalidCache if {@code true}, the catalog cache will be invalidated
+     *                     and reloaded during the refresh process.
+     */
+    public void resetToUninitialized(boolean invalidCache) {
         this.objectCreated = false;
         this.initialized = false;
         synchronized (this.propLock) {
@@ -511,6 +527,7 @@ public void onRefresh(boolean invalidCache) {
         synchronized (this.confLock) {
             this.cachedConf = null;
         }
+        onClose();
 
         refreshOnlyCatalogCache(invalidCache);
     }
@@ -727,7 +744,12 @@ public void onClose() {
         if (threadPoolWithPreAuth != null) {
             ThreadPoolManager.shutdownExecutorService(threadPoolWithPreAuth);
         }
-        CatalogIf.super.onClose();
+        if (null != preExecutionAuthenticator) {
+            preExecutionAuthenticator = null;
+        }
+        if (null != transactionManager) {
+            transactionManager = null;
+        }
     }
 
     private void removeAccessController() {

fe/fe-core/src/main/java/org/apache/doris/datasource/hive/HMSExternalCatalog.java

@@ -190,13 +190,24 @@ protected void initLocalObjectsImpl() {
     }
 
     @Override
-    public void onRefresh(boolean invalidCache) {
-        super.onRefresh(invalidCache);
+    public void resetToUninitialized(boolean invalidCache) {
+        super.resetToUninitialized(invalidCache);
         if (metadataOps != null) {
             metadataOps.close();
         }
     }
 
+    @Override
+    public void onClose() {
+        super.onClose();
+        if (null != fileSystemExecutor) {
+            ThreadPoolManager.shutdownExecutorService(fileSystemExecutor);
+        }
+        if (null != icebergMetadataOps) {
+            icebergMetadataOps.close();
+        }
+    }
+
     @Override
     public List<String> listTableNames(SessionContext ctx, String dbName) {
         makeSureInitialized();

fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalCatalog.java

@@ -98,6 +98,14 @@ public List<String> listTableNames(SessionContext ctx, String dbName) {
         return metadataOps.listTableNames(dbName);
     }
 
+    @Override
+    public void onClose() {
+        super.onClose();
+        if (null != catalog) {
+            catalog = null;
+        }
+    }
+
     protected void initS3Param(Configuration conf) {
         Map<String, String> properties = catalogProperty.getHadoopProperties();
         conf.set(Constants.AWS_CREDENTIALS_PROVIDER, PropertyConverter.getAWSCredentialsProviders(properties));

fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataOps.java

@@ -86,6 +86,9 @@ public ExternalCatalog getExternalCatalog() {
 
     @Override
     public void close() {
+        if (catalog != null) {
+            catalog = null;
+        }
     }
 
     @Override

fe/fe-core/src/main/java/org/apache/doris/datasource/jdbc/JdbcExternalCatalog.java

@@ -125,12 +125,8 @@ public void setDefaultPropsIfMissing(boolean isReplay) {
     }
 
     @Override
-    public void onRefresh(boolean invalidCache) {
-        super.onRefresh(invalidCache);
-        if (jdbcClient != null) {
-            jdbcClient.closeClient();
-            jdbcClient = null;
-        }
+    public void resetToUninitialized(boolean invalidCache) {
+        super.resetToUninitialized(invalidCache);
         this.identifierMapping = new JdbcIdentifierMapping(
                 (Env.isTableNamesCaseInsensitive() || Env.isStoredTableNamesLowerCase()),
                 Boolean.parseBoolean(getLowerCaseMetaNames()),

fe/fe-core/src/main/java/org/apache/doris/mysql/privilege/AccessControllerManager.java

@@ -34,6 +34,7 @@
 
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Maps;
+import org.apache.commons.lang3.StringUtils;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
@@ -167,7 +168,12 @@ private String getPluginIdentifierForAccessController(String acClassName) {
     }
 
     public void removeAccessController(String ctl) {
-        ctlToCtlAccessController.remove(ctl);
+        if (StringUtils.isBlank(ctl)) {
+            return;
+        }
+        if (ctlToCtlAccessController.containsKey(ctl)) {
+            ctlToCtlAccessController.remove(ctl);
+        }
         LOG.info("remove access controller for catalog {}", ctl);
     }