[core] Introduce endInput in StandardLineReader to avoid close input (#6574)

Author: JingsongLi

paimon-format/src/main/java/org/apache/paimon/format/text/StandardLineReader.java

@@ -43,14 +43,15 @@ public class StandardLineReader implements TextLineReader {
     private int bufferStart;
     private int bufferEnd;
     private int bufferPosition;
-    private boolean closed;
+    private boolean endInput;
 
     public StandardLineReader(InputStream in, long offset, @Nullable Long length)
             throws IOException {
         this.in = in;
         this.buffer = new byte[8192];
         this.length = length;
         this.lineBuilder = new ByteArrayOutputStream();
+        this.endInput = false;
         if (offset == 0) {
             readAtBeginning();
         } else {
@@ -80,16 +81,16 @@ private void readAtOffset(long offset) throws IOException {
     }
 
     private void skipFirstLine() throws IOException {
-        while (!closed) {
+        while (!endInput) {
             if (reachLengthLimit()) {
-                close();
+                endInput();
                 return;
             }
 
             // fill buffer if necessary
             if (bufferPosition >= bufferEnd) {
                 fillBuffer();
-                if (closed) {
+                if (endInput) {
                     return;
                 }
             }
@@ -107,15 +108,15 @@ public String readLine() throws IOException {
         lineBuilder.reset();
 
         if (reachLengthLimit()) {
-            close();
+            endInput();
             return null;
         }
 
         if (bufferPosition >= bufferEnd) {
             fillBuffer();
         }
 
-        while (!closed) {
+        while (!endInput) {
             if (seekToStartOfLineTerminator()) {
                 copyToLineBuilder();
                 seekPastLineTerminator();
@@ -158,7 +159,7 @@ private void seekPastLineTerminator() throws IOException {
             // fill buffer if necessary
             if (bufferPosition >= bufferEnd) {
                 fillBuffer();
-                if (closed) {
+                if (endInput) {
                     return;
                 }
             }
@@ -172,15 +173,15 @@ private void seekPastLineTerminator() throws IOException {
     }
 
     private void fillBuffer() throws IOException {
-        if (closed) {
+        if (endInput) {
             return;
         }
         checkArgument(bufferPosition >= bufferEnd, "Buffer is not empty");
         bufferStart = 0;
         bufferPosition = 0;
         bufferEnd = IOUtils.readNBytes(in, buffer, 0, buffer.length);
         if (bufferEnd == 0) {
-            close();
+            endInput();
         }
     }
 
@@ -205,9 +206,12 @@ private String buildLine() throws UnsupportedEncodingException {
         return lineBuilder.toString(UTF_8.name());
     }
 
+    private void endInput() {
+        endInput = true;
+    }
+
     @Override
     public void close() throws IOException {
-        closed = true;
         in.close();
     }
 }

paimon-format/src/test/java/org/apache/paimon/format/text/StandardLineReaderTest.java

@@ -31,6 +31,7 @@
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -56,6 +57,33 @@ public void tearDown() throws IOException {
         }
     }
 
+    /**
+     * After the stream is closed, we cannot use getPos, so we need to ensure that it is manually
+     * closed.
+     */
+    @Test
+    public void testCloseShouldNotBeInvoked() throws IOException {
+        String content = "line1\nline2\nline3";
+        AtomicBoolean closed = new AtomicBoolean(false);
+        try (TextLineReader reader =
+                TextLineReader.create(
+                        new ByteArrayInputStream(content.getBytes(StandardCharsets.UTF_8)) {
+                            @Override
+                            public void close() {
+                                closed.set(true);
+                            }
+                        },
+                        "\n",
+                        0,
+                        null)) {
+            assertThat(reader.readLine()).isEqualTo("line1");
+            assertThat(reader.readLine()).isEqualTo("line2");
+            assertThat(reader.readLine()).isEqualTo("line3");
+            assertThat(reader.readLine()).isNull();
+            assertThat(closed.get()).isFalse();
+        }
+    }
+
     @Test
     public void testReadLineWithLineFeedDelimiter() throws IOException {
         // Test basic functionality with \n delimiter