[core] Fix file operation thread limit

paimon-common/src/main/java/org/apache/paimon/utils/FileOperationThreadPool.java

@@ -20,6 +20,7 @@
 
 import org.apache.paimon.fs.FileIO;
 
+import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ThreadPoolExecutor;
 
 import static org.apache.paimon.utils.ThreadPoolUtils.createCachedThreadPool;
@@ -32,10 +33,13 @@ public class FileOperationThreadPool {
     private static ThreadPoolExecutor executorService =
             createCachedThreadPool(Runtime.getRuntime().availableProcessors(), THREAD_NAME);
 
-    public static synchronized ThreadPoolExecutor getExecutorService(int threadNum) {
-        if (threadNum <= executorService.getMaximumPoolSize()) {
+    public static synchronized ExecutorService getExecutorService(int threadNum) {
+        if (threadNum <= 0 || threadNum == executorService.getMaximumPoolSize()) {
             return executorService;
         }
+        if (threadNum < executorService.getMaximumPoolSize()) {
+            return new SemaphoredDelegatingExecutor(executorService, threadNum, false);
+        }
         // we don't need to close previous pool
         // it is just cached pool
         executorService = createCachedThreadPool(threadNum, THREAD_NAME);

paimon-core/src/main/java/org/apache/paimon/operation/ListUnexistingFiles.java

@@ -38,14 +38,14 @@
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
-import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.ExecutorService;
 
 /** List what data files recorded in manifests are missing from the filesystem. */
 public class ListUnexistingFiles {
 
     private final FileStoreTable table;
     private final FileStorePathFactory pathFactory;
-    private final ThreadPoolExecutor executor;
+    private final ExecutorService executor;
 
     public ListUnexistingFiles(FileStoreTable table) {
         this.table = table;

paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java

@@ -61,7 +61,6 @@
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import java.util.function.Predicate;
@@ -89,7 +88,7 @@ public class TableCommitImpl implements InnerTableCommit {
     private final AtomicReference<Throwable> maintainError;
     private final String tableName;
     private final boolean forceCreatingSnapshot;
-    private final ThreadPoolExecutor fileCheckExecutor;
+    private final ExecutorService fileCheckExecutor;
 
     @Nullable private Map<String, String> overwritePartitionSpec = null;
     @Nullable private List<BinaryRow> overwriteStaticPartitions = null;