From 894248f57e6ecd9ebe10eed84d081c8a9160d335 Mon Sep 17 00:00:00 2001
From: ZhongYujiang <42907416+zhongyujiang@users.noreply.github.com>
Date: Tue, 10 Dec 2024 14:55:00 +0800
Subject: [PATCH] [core]: Improve fault tolerance for data spill to disk.

---
 .../paimon/disk/FileChannelManagerImpl.java   | 22 ++++++++++++++-----
 1 file changed, 16 insertions(+), 6 deletions(-)

diff --git a/paimon-core/src/main/java/org/apache/paimon/disk/FileChannelManagerImpl.java b/paimon-core/src/main/java/org/apache/paimon/disk/FileChannelManagerImpl.java
index ce175e90bbd1..99690d426fb5 100644
--- a/paimon-core/src/main/java/org/apache/paimon/disk/FileChannelManagerImpl.java
+++ b/paimon-core/src/main/java/org/apache/paimon/disk/FileChannelManagerImpl.java
@@ -29,7 +29,9 @@
 import java.io.File;
 import java.io.IOException;
 import java.io.UncheckedIOException;
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.List;
 import java.util.Random;
 import java.util.UUID;
 import java.util.concurrent.atomic.AtomicLong;
@@ -63,24 +65,32 @@ public FileChannelManagerImpl(String[] tempDirs, String prefix) {
     }
 
     private static File[] createFiles(String[] tempDirs, String prefix) {
-        File[] files = new File[tempDirs.length];
+        List<File> filesList = new ArrayList<>();
         for (int i = 0; i < tempDirs.length; i++) {
             File baseDir = new File(tempDirs[i]);
             String subfolder = String.format("paimon-%s-%s", prefix, UUID.randomUUID());
             File storageDir = new File(baseDir, subfolder);
 
             if (!storageDir.exists() && !storageDir.mkdirs()) {
-                throw new RuntimeException(
-                        "Could not create storage directory for FileChannelManager: "
-                                + storageDir.getAbsolutePath());
+                LOG.warn(
+                        "Failed to create directory {}, temp directory {} will not be used",
+                        storageDir.getAbsolutePath(),
+                        tempDirs[i]);
+                continue;
             }
-            files[i] = storageDir;
+
+            filesList.add(storageDir);
 
             LOG.debug(
                     "FileChannelManager uses directory {} for spill files.",
                     storageDir.getAbsolutePath());
         }
-        return files;
+
+        if (filesList.isEmpty()) {
+            throw new RuntimeException("No available temporary directories");
+        }
+
+        return filesList.toArray(new File[0]);
     }
 
     @Override