[common] Close input when copy output close fails

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

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");
+        }
+    }
 }