From 402a8bc00547e9b179d012649939dd08ae8a5681 Mon Sep 17 00:00:00 2001
From: wolfkill <27876473+wolfkill@users.noreply.github.com>
Date: Thu, 23 Jul 2026 09:47:56 +0800
Subject: [PATCH] [common] Close input when copy output close fails

---
 .../java/org/apache/fluss/utils/IOUtils.java  | 19 +++++++++-
 .../org/apache/fluss/utils/IOUtilsTest.java   | 38 +++++++++++++++++++
 2 files changed, 55 insertions(+), 2 deletions(-)

diff --git a/fluss-common/src/main/java/org/apache/fluss/utils/IOUtils.java b/fluss-common/src/main/java/org/apache/fluss/utils/IOUtils.java
index 0b268eacf2..da291c0ec7 100644
--- a/fluss-common/src/main/java/org/apache/fluss/utils/IOUtils.java
+++ b/fluss-common/src/main/java/org/apache/fluss/utils/IOUtils.java
@@ -73,8 +73,23 @@ private static long copyBytes(
             return totalBytes;
         } finally {
             if (close) {
-                out.close();
-                in.close();
+                Throwable closeException = null;
+                try {
+                    out.close();
+                } catch (Throwable t) {
+                    closeException = t;
+                }
+                try {
+                    in.close();
+                } catch (Throwable t) {
+                    closeException = ExceptionUtils.firstOrSuppressed(t, closeException);
+                }
+                if (closeException != null) {
+                    if (closeException instanceof IOException) {
+                        throw (IOException) closeException;
+                    }
+                    ExceptionUtils.rethrow(closeException);
+                }
             }
         }
     }
diff --git a/fluss-common/src/test/java/org/apache/fluss/utils/IOUtilsTest.java b/fluss-common/src/test/java/org/apache/fluss/utils/IOUtilsTest.java
index f50987cd77..aeca3c214c 100644
--- a/fluss-common/src/test/java/org/apache/fluss/utils/IOUtilsTest.java
+++ b/fluss-common/src/test/java/org/apache/fluss/utils/IOUtilsTest.java
@@ -22,6 +22,7 @@
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.io.OutputStream;
 import java.util.Arrays;
 
 import static org.assertj.core.api.Assertions.assertThat;
@@ -74,6 +75,17 @@ void testCopyBytes() throws IOException {
         assertThat(buf).isEqualTo(new byte[] {'h', 'e', 'l', 'l', 'o'});
     }
 
+    @Test
+    void testCopyBytesClosesInputWhenOutputCloseFails() {
+        TrackingInputStream in = new TrackingInputStream("hello".getBytes());
+        OutputStream out = new CloseFailingOutputStream();
+
+        assertThatThrownBy(() -> IOUtils.copyBytes(in, out, true))
+                .isInstanceOf(IOException.class)
+                .hasMessage("Fail to close output");
+        assertThat(in.closed).isTrue();
+    }
+
     @Test
     void testReadFully() throws IOException {
         ByteArrayOutputStream out = new ByteArrayOutputStream();
@@ -108,4 +120,30 @@ public void close() throws Exception {
             closed = true;
         }
     }
+
+    private static class TrackingInputStream extends ByteArrayInputStream {
+
+        private boolean closed;
+
+        private TrackingInputStream(byte[] buf) {
+            super(buf);
+        }
+
+        @Override
+        public void close() throws IOException {
+            closed = true;
+            super.close();
+        }
+    }
+
+    private static class CloseFailingOutputStream extends OutputStream {
+
+        @Override
+        public void write(int b) {}
+
+        @Override
+        public void close() throws IOException {
+            throw new IOException("Fail to close output");
+        }
+    }
 }