branch-4.0: [Fix](catalog)Resources should be closed when dropping a Catalog.

apache#59512
apache#59063

Author: CalvinKirs

.licenserc.yaml

@@ -43,6 +43,7 @@ header:
     - "**/*.parquet"
     - "docs/.markdownlintignore"
     - "fe/fe-core/src/test/resources/data/net_snmp_normal"
+    - "fe/fe-core/src/main/java/software/amazon/awssdk/core/client/builder/SdkDefaultClientBuilder.java"
     - "fe/fe-core/src/main/antlr4/org/apache/doris/nereids/JavaLexer.g4"
     - "fe/fe-core/src/main/antlr4/org/apache/doris/nereids/JavaParser.g4"
     - "be/dict/ik/*"

LICENSE.txt

@@ -202,6 +202,13 @@
 
 --------------------------------------------------------------------------------
 
+The following components are provided under the Apache License. See project link for details.
+The text of each license is the standard Apache 2.0 license.
+
+software.amazon.awssdk.core.client.builder.SdkDefaultClientBuilder from AWS SDK v2 (sdk-core 2.29.52)
+
+--------------------------------------------------------------------------------
+
 be/src/common/status.* : BSD-style license
 
   Copyright (c) 2011 The LevelDB Authors. All rights reserved.

NOTICE.txt

@@ -52,3 +52,33 @@ its NOTICE file:
     This product includes cryptographic software written by Eric Young
     (eay@cryptsoft.com).  This product includes software written by Tim
     Hudson (tjh@cryptsoft.com).
+
+--------------------------------------------------------------------------------    
+This product includes code from AWS SDK, which includes the following in
+its NOTICE file:
+ 
+AWS SDK for Java 2.0
+Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
+
+This product includes software developed by
+Amazon Technologies, Inc (http://www.amazon.com/).
+
+**********************
+THIRD PARTY COMPONENTS
+**********************
+This software includes third party software subject to the following copyrights:
+- XML parsing and utility functions from JetS3t - Copyright 2006-2009 James Murty.
+- PKCS#1 PEM encoded private key parsing and utility functions from oauth.googlecode.com - Copyright 1998-2010 AOL Inc.
+- Apache Commons Lang - https://github.com/apache/commons-lang
+- Netty Reactive Streams - https://github.com/playframework/netty-reactive-streams
+- Jackson-core - https://github.com/FasterXML/jackson-core
+- Jackson-dataformat-cbor - https://github.com/FasterXML/jackson-dataformats-binary
+
+The licenses for these third party components are included in LICENSE.txt
+
+- For Apache Commons Lang see also this required NOTICE:
+  Apache Commons Lang
+  Copyright 2001-2020 The Apache Software Foundation
+  
+  This product includes software developed at
+  The Apache Software Foundation (https://www.apache.org/).    

fe/check/checkstyle/suppressions.xml

@@ -74,4 +74,5 @@ under the License.
 
     <!-- ignore gensrc/thrift/ExternalTableSchema.thrift -->
     <suppress files=".*thrift/schema/external/.*" checks=".*"/>
+    <suppress files="software/amazon/awssdk/core/client/builder/SdkDefaultClientBuilder.java" checks=".*"/>
 </suppressions>

fe/fe-common/src/main/java/org/apache/doris/common/Config.java

@@ -3729,6 +3729,13 @@ public static int metaServiceRpcRetryTimes() {
     @ConfField(mutable = true)
     public static String aws_credentials_provider_version = "v2";
 
+    @ConfField(description = {
+            "AWS SDK 用于调度异步重试、超时任务以及其他后台操作的线程池大小，全局共享",
+            "The thread pool size used by the AWS SDK to schedule asynchronous retries, timeout tasks, "
+                    + "and other background operations, shared globally"
+    })
+    public static int aws_sdk_async_scheduler_thread_pool_size = 20;
+
     @ConfField(description = {
             "agent tasks 健康检查的时间间隔，默认五分钟，小于等于0时不做健康检查",
             "agent tasks health check interval, default is five minutes, no health check when less than or equal to 0"

fe/fe-core/pom.xml

@@ -636,14 +636,6 @@ under the License.
             </exclusions>
         </dependency>
 
-        <!-- for fe recognize files stored on gcs -->
-        <dependency>
-            <groupId>com.google.cloud.bigdataoss</groupId>
-            <artifactId>gcs-connector</artifactId>
-            <version>${gcs.version}</version>
-            <scope>${gcs.dependency.scope}</scope>
-        </dependency>
-
         <!-- https://mvnrepository.com/artifact/org.apache.ranger/ranger-plugins-common -->
         <dependency>
             <groupId>org.apache.ranger</groupId>

fe/fe-core/src/main/java/org/apache/doris/common/util/UncloseableScheduledExecutorService.java

@@ -0,0 +1,153 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.common.util;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+public final class UncloseableScheduledExecutorService
+        implements ScheduledExecutorService {
+
+    private final ScheduledExecutorService delegate;
+
+    public UncloseableScheduledExecutorService(
+            ScheduledExecutorService delegate) {
+        this.delegate = Objects.requireNonNull(delegate);
+    }
+
+    // ================= Lifecycle methods (NO-OP) =================
+    @Override
+    public void shutdown() {
+        // NO-OP
+    }
+
+    @Override
+    public List<Runnable> shutdownNow() {
+        return Collections.emptyList();
+    }
+
+    @Override
+    public boolean isShutdown() {
+        return false;
+    }
+
+    @Override
+    public boolean isTerminated() {
+        return false;
+    }
+
+    @Override
+    public boolean awaitTermination(long timeout, TimeUnit unit) {
+        return false;
+    }
+
+    // ================= Scheduled methods =================
+    @Override
+    public ScheduledFuture<?> schedule(
+            Runnable command, long delay, TimeUnit unit) {
+        return delegate.schedule(command, delay, unit);
+    }
+
+    @Override
+    public <V> ScheduledFuture<V> schedule(
+            Callable<V> callable, long delay, TimeUnit unit) {
+        return delegate.schedule(callable, delay, unit);
+    }
+
+    @Override
+    public ScheduledFuture<?> scheduleAtFixedRate(
+            Runnable command,
+            long initialDelay,
+            long period,
+            TimeUnit unit) {
+        return delegate.scheduleAtFixedRate(
+                command, initialDelay, period, unit);
+    }
+
+    @Override
+    public ScheduledFuture<?> scheduleWithFixedDelay(
+            Runnable command,
+            long initialDelay,
+            long delay,
+            TimeUnit unit) {
+        return delegate.scheduleWithFixedDelay(
+                command, initialDelay, delay, unit);
+    }
+
+    // ================= Executor methods =================
+    @Override
+    public void execute(Runnable command) {
+        delegate.execute(command);
+    }
+
+    @Override
+    public <T> Future<T> submit(Callable<T> task) {
+        return delegate.submit(task);
+    }
+
+    @Override
+    public <T> Future<T> submit(Runnable task, T result) {
+        return delegate.submit(task, result);
+    }
+
+    @Override
+    public Future<?> submit(Runnable task) {
+        return delegate.submit(task);
+    }
+
+    @Override
+    public <T> List<Future<T>> invokeAll(
+            Collection<? extends Callable<T>> tasks)
+            throws InterruptedException {
+        return delegate.invokeAll(tasks);
+    }
+
+    @Override
+    public <T> List<Future<T>> invokeAll(
+            Collection<? extends Callable<T>> tasks,
+            long timeout,
+            TimeUnit unit)
+            throws InterruptedException {
+        return delegate.invokeAll(tasks, timeout, unit);
+    }
+
+    @Override
+    public <T> T invokeAny(
+            Collection<? extends Callable<T>> tasks)
+            throws InterruptedException, ExecutionException {
+        return delegate.invokeAny(tasks);
+    }
+
+    @Override
+    public <T> T invokeAny(
+            Collection<? extends Callable<T>> tasks,
+            long timeout,
+            TimeUnit unit)
+            throws InterruptedException, ExecutionException, TimeoutException {
+        return delegate.invokeAny(tasks, timeout, unit);
+    }
+}

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);