[Fix](catalog)Resources should be closed when dropping a Catalog.

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

@@ -34,13 +34,16 @@
 
 import org.apache.commons.lang3.math.NumberUtils;
 import org.apache.iceberg.catalog.Catalog;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
 
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
 
 public abstract class IcebergExternalCatalog extends ExternalCatalog {
 
+    private static final Logger LOG = LogManager.getLogger(IcebergExternalCatalog.class);
     public static final String ICEBERG_CATALOG_TYPE = "iceberg.catalog.type";
     public static final String ICEBERG_REST = "rest";
     public static final String ICEBERG_HMS = "hms";
@@ -209,7 +212,14 @@ public List<String> listTableNames(SessionContext ctx, String dbName) {
     public void onClose() {
         super.onClose();
         if (null != catalog) {
-            catalog = null;
+            try {
+                if (catalog instanceof AutoCloseable) {
+                    ((AutoCloseable) catalog).close();
+                }
+                catalog = null;
+            } catch (Exception e) {
+                LOG.warn("Failed to close iceberg catalog: {}", getName(), e);
+            }
         }
     }
 

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

@@ -190,4 +190,16 @@ public void checkProperties() throws DdlException {
         super.checkProperties();
         catalogProperty.checkMetaStoreAndStorageProperties(AbstractPaimonProperties.class);
     }
+
+    @Override
+    public void onClose() {
+        super.onClose();
+        if (null != catalog) {
+            try {
+                catalog.close();
+            } catch (Exception e) {
+                LOG.warn("Failed to close paimon catalog: {}", getName(), e);
+            }
+        }
+    }
 }

fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/AbstractIcebergProperties.java

@@ -111,7 +111,8 @@ protected AbstractIcebergProperties(Map<String, String> props) {
      * This field is used to perform metadata operations like creating, querying,
      * and deleting Iceberg tables.
      */
-    public final Catalog initializeCatalog(String catalogName, List<StorageProperties> storagePropertiesList) {
+    public final Catalog initializeCatalog(String catalogName,
+                                                        List<StorageProperties> storagePropertiesList) {
         Map<String, String> catalogProps = new HashMap<>(getOrigProps());
         if (StringUtils.isNotBlank(warehouse)) {
             catalogProps.put(CatalogProperties.WAREHOUSE_LOCATION, warehouse);

fe/fe-core/src/test/java/org/apache/doris/datasource/property/metastore/AbstractIcebergPropertiesTest.java

@@ -48,8 +48,8 @@ public String getIcebergCatalogType() {
 
         @Override
         protected Catalog initCatalog(String catalogName,
-                                      Map<String, String> catalogProps,
-                                      List<StorageProperties> storagePropertiesList) {
+                                                   Map<String, String> catalogProps,
+                                                   List<StorageProperties> storagePropertiesList) {
             // Capture the catalogProps for verification
             this.capturedCatalogProps = new HashMap<>(catalogProps);
             return catalogToReturn;

regression-test/suites/external_table_p2/paimon/test_paimon_dlf_catalog.groovy

@@ -64,6 +64,8 @@ suite("test_paimon_dlf_catalog", "p2,external,paimon,external_remote,external_re
         sql """SELECT * FROM tb_simple\$snapshots;"""
     } finally {
         sql """set force_jni_scanner=false"""
+
+        sql """drop catalog if exists ${catalog};"""
     }
 }
 

regression-test/suites/external_table_p2/paimon/test_paimon_dlf_catalog_miss_dlf_param.groovy

@@ -69,6 +69,7 @@ suite("test_paimon_dlf_catalog_miss_dlf_param", "p2,external,paimon,external_rem
 
     } finally {
         sql """set force_jni_scanner=false"""
+        sql """drop catalog if exists ${catalog};"""
     }
 }
 

regression-test/suites/external_table_p2/paimon/test_paimon_dlf_catalog_new_param.groovy

@@ -71,6 +71,8 @@ suite("test_paimon_dlf_catalog_new_param", "p2,external,paimon,external_remote,e
 
     } finally {
         sql """set force_jni_scanner=false"""
+
+        sql """drop catalog if exists ${catalog};"""
     }
 }
 

regression-test/suites/external_table_p2/paimon/test_paimon_dlf_rest_catalog.groovy

@@ -44,6 +44,8 @@ suite("test_paimon_dlf_rest_catalog", "p2,external,paimon,external_remote,extern
 
     } finally {
         sql """set force_jni_scanner=false"""
+
+        sql """drop catalog if exists ${catalog};"""
     }
 }
 

regression-test/suites/external_table_p2/paimon/test_paimon_hms_catalog.groovy

@@ -56,6 +56,9 @@ suite("test_paimon_hms_catalog", "p2,external,paimon,new_catalog_property") {
         // TODO(zgx): add branch/tag
         // system table
         sql """SELECT * FROM external_test_table\$snapshots;"""
+
+
+        sql """drop catalog if exists ${catalog};"""
     }
     String enabled = context.config.otherConfigs.get("enablePaimonTest")
     if (enabled == null || !enabled.equalsIgnoreCase("true")) {

regression-test/suites/external_table_p2/refactor_catalog_param/hive_on_hms_and_dlf.groovy

@@ -81,6 +81,10 @@ suite("hive_on_hms_and_dlf", "p2,external,new_catalog_property") {
             show databases  like "${db_name}";
         """
         assert dropResult.size() == 0
+
+        sql """
+            DROP CATALOG IF EXISTS ${catalog_name};
+        """
     }
 
     /*--------test partition table insert---------*/
@@ -181,6 +185,10 @@ suite("hive_on_hms_and_dlf", "p2,external,new_catalog_property") {
             show databases  like "${db_name}";
         """
         assert dropResult.size() == 0
+
+        sql """
+            DROP CATALOG IF EXISTS ${catalog_name};
+        """
     }
 
     /*--------test insert overwrite---------*/
@@ -278,6 +286,10 @@ suite("hive_on_hms_and_dlf", "p2,external,new_catalog_property") {
             show databases  like "${db_name}";
         """
         assert dropResult.size() == 0
+
+        sql """
+            DROP CATALOG IF EXISTS ${catalog_name};
+        """
     }
 
     /*--------only execute query---------*/
@@ -311,6 +323,10 @@ suite("hive_on_hms_and_dlf", "p2,external,new_catalog_property") {
             SELECT count(1) FROM ${table_name};
         """
         assert queryResult.get(0).get(0) == data_count
+
+        sql """
+            DROP CATALOG IF EXISTS ${catalog_name};
+        """
     }
     String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")
     String keytab_root_dir = "/keytabs"

regression-test/suites/external_table_p2/refactor_catalog_param/iceberg_and_hive_on_glue.groovy

@@ -80,6 +80,9 @@ suite("iceberg_and_hive_on_glue", "p2,external,hive,new_catalog_property") {
             show databases  like "${db_name}";
         """
         assert dropResult.size() == 0
+        sql """
+            DROP CATALOG IF EXISTS ${catalog_name};
+        """
     }
 
     def testQueryAndInsertIcerberg = { String catalogProperties, String prefix ->
@@ -141,6 +144,9 @@ suite("iceberg_and_hive_on_glue", "p2,external,hive,new_catalog_property") {
             show databases  like "${db_name}";
         """
         assert dropResult.size() == 0
+        sql """
+            DROP CATALOG IF EXISTS ${catalog_name};
+        """
     }
 
     /*--------test insert overwrite for hive---------*/
@@ -229,6 +235,9 @@ suite("iceberg_and_hive_on_glue", "p2,external,hive,new_catalog_property") {
             show databases  like "${db_name}";
         """
         assert dropResult.size() == 0
+        sql """
+            DROP CATALOG IF EXISTS ${catalog_name};
+        """
     }
 
     /*--------test insert overwrite for iceberg---------*/
@@ -313,6 +322,10 @@ suite("iceberg_and_hive_on_glue", "p2,external,hive,new_catalog_property") {
             show databases  like "${db_name}";
         """
         assert dropResult.size() == 0
+
+        sql """
+            DROP CATALOG IF EXISTS ${catalog_name};
+        """
     }
 
     /*--------only execute query---------*/
@@ -346,6 +359,10 @@ suite("iceberg_and_hive_on_glue", "p2,external,hive,new_catalog_property") {
             SELECT count(1) FROM ${table_name};
         """
         assert queryResult.get(0).get(0) == data_count
+
+        sql """
+            DROP CATALOG IF EXISTS ${catalog_name};
+        """
     }
 
     /*--------GLUE START-----------*/

regression-test/suites/external_table_p2/refactor_catalog_param/iceberg_on_hms_and_filesystem_and_dlf.groovy

@@ -178,6 +178,9 @@ suite("iceberg_on_hms_and_filesystem_and_dlf", "p2,external,new_catalog_property
             show databases  like "${db_name}";
         """
         assert dropResult.size() == 0
+        sql """
+            DROP CATALOG IF EXISTS ${catalog_name};
+        """
     }
 
     /*--------only execute query---------*/
@@ -211,6 +214,10 @@ suite("iceberg_on_hms_and_filesystem_and_dlf", "p2,external,new_catalog_property
             SELECT count(1) FROM ${table_name};
         """
         assert queryResult.get(0).get(0) == data_count
+
+        sql """
+            DROP CATALOG IF EXISTS ${catalog_name};
+        """
     }
 
     String externalEnvIp = context.config.otherConfigs.get("externalEnvIp")