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

### What problem does this PR solve?
When creating an AWS SDK V2 async client (e.g. S3AsyncClient), the SDK
requires a thread pool to manage asynchronous task scheduling, timeouts,
and retries (e.g. ScheduledExecutorService or async executor).


If no executor is explicitly provided, the AWS SDK creates its own
internal thread pool, which is expected to be shut down when
client.close() is invoked.

### Issue

In Doris, when using Paimon Catalog, some catalog implementations
provide an empty close() method. As a result, when a user executes DROP
CATALOG:

starting with Hadoop 3.4, the AWS SDK was upgraded to v2. Since #57226
upgraded HDFS to 3.4.2, the catalog runs into this issue.

Some Catalog instance is discarded,AWS SDK client.close() is never
called,The internally created thread pool cannot be shut down.

This leads to a thread leak

FYI aws/aws-sdk-java-v2#3746

### Problem Analysis

The lifecycle of Catalog creation and destruction is complex and managed
internally by Paimon

Doris cannot reliably intervene in the call chain to enforce AWS SDK
client.close()

If each Catalog creates its own AWS SDK client with an internally
managed thread pool, threads will continue to leak as Catalogs are
repeatedly created and dropped

 ### Solution

This PR resolves the issue by introducing a shared executor strategy:

A Doris-managed shared thread pool is explicitly passed when creating
AWS SDK V2 clients


This prevents the AWS SDK from implicitly creating per-client internal
thread pools

The lifecycle of the executor is fully controlled by Doris and no longer
depends on Paimon Catalog’s close() implementation


With this approach, even if a Paimon Catalog’s close() method is a
no-op, the system will no longer leak threads.
None

Author: CalvinKirs

.licenserc.yaml

@@ -44,6 +44,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,6 @@ 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=".*"/>
+    <suppress files="software/amazon/awssdk/core/client/builder/UncloseableScheduledExecutorService.java" checks=".*"/>
 </suppressions>

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

@@ -3771,6 +3771,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/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";
@@ -210,7 +213,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);