[core] AsyncPositionOutputStream should not flush after close. (#3967)

apache · Aug 15, 2024 · 9f3389e · 9f3389e
1 parent 0af4feb
commit 9f3389e
Show file tree

Hide file tree

Showing 2 changed files with 17 additions and 0 deletions.
diff --git a/paimon-common/src/main/java/org/apache/paimon/fs/AsyncPositionOutputStream.java b/paimon-common/src/main/java/org/apache/paimon/fs/AsyncPositionOutputStream.java
@@ -52,6 +52,7 @@ public class AsyncPositionOutputStream extends PositionOutputStream {
 
     private int totalBuffers;
     private long position;
+    private boolean closed = false;
 
     public AsyncPositionOutputStream(PositionOutputStream out) {
         this.out = out;
@@ -172,6 +173,9 @@ public void write(byte[] b, int off, int len) throws IOException {
 
     @Override
     public void flush() throws IOException {
+        if (closed) {
+            throw new IOException("Already closed");
+        }
         checkException();
         flushBuffer();
         FlushEvent event = new FlushEvent();
@@ -193,6 +197,9 @@ public void flush() throws IOException {
 
     @Override
     public void close() throws IOException {
+        if (closed) {
+            return;
+        }
         checkException();
         flushBuffer();
         sendEndEvent();
@@ -204,6 +211,7 @@ public void close() throws IOException {
         } catch (ExecutionException e) {
             throw new RuntimeException(e);
         }
+        this.closed = true;
     }
 
     private void sendEndEvent() {

diff --git a/paimon-common/src/test/java/org/apache/paimon/fs/AsyncPositionOutputStreamTest.java b/paimon-common/src/test/java/org/apache/paimon/fs/AsyncPositionOutputStreamTest.java
@@ -25,6 +25,7 @@
 import java.util.concurrent.ThreadLocalRandom;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 class AsyncPositionOutputStreamTest {
@@ -173,6 +174,14 @@ public void write(byte[] b, int off, int len) {
         assertThat(out.closed).isTrue();
     }
 
+    @Test
+    public void testCloseFlushThrowsException() throws Exception {
+        AsyncPositionOutputStream asyncPositionOutputStream =
+                new AsyncPositionOutputStream(new ByteArrayPositionOutputStream());
+        asyncPositionOutputStream.close();
+        assertThatCode(asyncPositionOutputStream::flush).hasMessage("Already closed");
+    }
+
     private static class ByteArrayPositionOutputStream extends PositionOutputStream {
 
         private final ByteArrayOutputStream out;