From 2e345833480b1660750a92e4042d3f6b332bd50e Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Wed, 15 Jul 2026 07:02:18 -0700 Subject: [PATCH 1/2] [fix][io] File source: read zip entries so zip files emit records ZipFiles.lines() wrapped a ZipInputStream in a reader but never called getNextEntry(). A ZipInputStream yields no data until it is positioned onto an entry, so the reader saw an empty stream and the File source emitted zero records for every zip file, then deleted/renamed it as if processed. Read through each file entry and return the collected lines (in entry order), so single- and multi-entry archives both deliver their content. Also make ZipFilesTest non-vacuous: streamZipFileTest previously asserted inside a forEachOrdered that ran zero times and swallowed exceptions, so it passed even though no lines were read. It now asserts the exact nine lines, and a new streamMultiEntryZipFileTest builds a two-entry archive and checks all lines are returned. Fixes #114 --- .../apache/pulsar/io/file/utils/ZipFiles.java | 50 +++++++++++-------- .../pulsar/io/file/utils/ZipFilesTest.java | 41 +++++++++++++-- 2 files changed, 66 insertions(+), 25 deletions(-) diff --git a/file/src/main/java/org/apache/pulsar/io/file/utils/ZipFiles.java b/file/src/main/java/org/apache/pulsar/io/file/utils/ZipFiles.java index bc4cfc3ffb..848d611849 100644 --- a/file/src/main/java/org/apache/pulsar/io/file/utils/ZipFiles.java +++ b/file/src/main/java/org/apache/pulsar/io/file/utils/ZipFiles.java @@ -20,7 +20,6 @@ import java.io.BufferedInputStream; import java.io.BufferedReader; -import java.io.Closeable; import java.io.DataInputStream; import java.io.File; import java.io.FileInputStream; @@ -29,7 +28,10 @@ import java.io.UncheckedIOException; import java.nio.file.Files; import java.nio.file.Path; +import java.util.ArrayList; +import java.util.List; import java.util.stream.Stream; +import java.util.zip.ZipEntry; import java.util.zip.ZipInputStream; /** @@ -51,34 +53,38 @@ public static boolean isZip(File f) { } /** - * Get a lazily loaded stream of lines from a gzipped file, similar to + * Get a stream of lines from every file entry of a zip file, similar to * {@link Files#lines(java.nio.file.Path)}. * * @param path * The path to the zipped file. - * @return stream with lines. + * @return stream with the lines of all file entries, in entry order. */ public static Stream lines(Path path) { - ZipInputStream zipStream = null; - - try { - zipStream = new ZipInputStream(Files.newInputStream(path)); + // A ZipInputStream returns no data until it is positioned onto an entry via + // getNextEntry(); without that call the reader sees an empty stream and no lines are + // produced. Read through every file entry so multi-entry archives contribute all of + // their lines, then return the collected lines as a stream (the archive must be fully + // read to walk its entries, so this cannot be lazy the way GZipFiles.lines is). + List lines = new ArrayList<>(); + try (ZipInputStream zipStream = new ZipInputStream(Files.newInputStream(path))) { + ZipEntry entry; + while ((entry = zipStream.getNextEntry()) != null) { + if (entry.isDirectory()) { + continue; + } + // Read the current entry to its end. read() returns -1 at the entry boundary, + // so the reader stops before the next entry; do not close it, as that would + // close the shared zipStream mid-iteration. + BufferedReader reader = new BufferedReader(new InputStreamReader(zipStream)); + String line; + while ((line = reader.readLine()) != null) { + lines.add(line); + } + } } catch (IOException e) { - closeSafely(zipStream); - throw new UncheckedIOException(e); - } - // Reader decoder = new InputStreamReader(gzipStream, Charset.defaultCharset()); - BufferedReader reader = new BufferedReader(new InputStreamReader(zipStream)); - return reader.lines().onClose(() -> closeSafely(reader)); - } - - private static void closeSafely(Closeable closeable) { - if (closeable != null) { - try { - closeable.close(); - } catch (IOException e) { - // Ignore - } + throw new UncheckedIOException(e); } + return lines.stream(); } } diff --git a/file/src/test/java/org/apache/pulsar/io/file/utils/ZipFilesTest.java b/file/src/test/java/org/apache/pulsar/io/file/utils/ZipFilesTest.java index 64f3ff42c7..99f85f1f7e 100644 --- a/file/src/test/java/org/apache/pulsar/io/file/utils/ZipFilesTest.java +++ b/file/src/test/java/org/apache/pulsar/io/file/utils/ZipFilesTest.java @@ -18,12 +18,19 @@ */ package org.apache.pulsar.io.file.utils; +import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertFalse; import static org.testng.Assert.assertTrue; import java.io.File; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; import java.nio.file.Path; import java.nio.file.Paths; +import java.util.List; +import java.util.stream.Collectors; import java.util.stream.Stream; +import java.util.zip.ZipEntry; +import java.util.zip.ZipOutputStream; import org.testng.annotations.Test; public class ZipFilesTest { @@ -52,10 +59,38 @@ public final void nonExistantGzipFileTest() { public final void streamZipFileTest() { Path path = Paths.get(getFile("org/apache/pulsar/io/file/validZip.zip").getAbsolutePath(), ""); + // validZip.zip contains a single entry with the nine lines "Line 1".."Line 9". try (Stream lines = ZipFiles.lines(path)) { - lines.forEachOrdered(line -> assertTrue(line.startsWith("Line "))); - } catch (Exception e) { - e.printStackTrace(); + List collected = lines.collect(Collectors.toList()); + assertEquals(collected.size(), 9, "expected nine lines from the zip entry"); + for (int i = 0; i < collected.size(); i++) { + assertEquals(collected.get(i), "Line " + (i + 1)); + } + } + } + + @Test + public final void streamMultiEntryZipFileTest() throws Exception { + // Build a two-entry archive in a temp file so the test is self-contained and proves + // that lines from every entry are returned, in entry order. + Path zip = Files.createTempFile("pulsar-io-file-ziptest", ".zip"); + try { + try (ZipOutputStream out = new ZipOutputStream(Files.newOutputStream(zip))) { + out.putNextEntry(new ZipEntry("first.txt")); + out.write("a1\na2".getBytes(StandardCharsets.UTF_8)); + out.closeEntry(); + out.putNextEntry(new ZipEntry("second.txt")); + out.write("b1\nb2\nb3".getBytes(StandardCharsets.UTF_8)); + out.closeEntry(); + } + + assertTrue(ZipFiles.isZip(zip.toFile())); + try (Stream lines = ZipFiles.lines(zip)) { + assertEquals(lines.collect(Collectors.toList()), + List.of("a1", "a2", "b1", "b2", "b3")); + } + } finally { + Files.deleteIfExists(zip); } } From 1e0730f95afd754d94a82e4020c8b6ebfadba38e Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Wed, 15 Jul 2026 07:42:29 -0700 Subject: [PATCH 2/2] [fix][io] Stream zip entries lazily instead of buffering all lines Address review feedback: ZipFiles.lines() collected every line into a List before returning a stream, which buffers the whole decompressed archive in memory and defeats the line-by-line back-pressure that FileConsumerThread relies on. Walk the entries lazily via a line iterator that opens a fresh reader per entry (preserving entry boundaries as line boundaries) and closes the ZipInputStream when the returned stream is closed, matching the lazy GZipFiles.lines contract. --- .../apache/pulsar/io/file/utils/ZipFiles.java | 118 ++++++++++++++---- 1 file changed, 93 insertions(+), 25 deletions(-) diff --git a/file/src/main/java/org/apache/pulsar/io/file/utils/ZipFiles.java b/file/src/main/java/org/apache/pulsar/io/file/utils/ZipFiles.java index 848d611849..13e729f1df 100644 --- a/file/src/main/java/org/apache/pulsar/io/file/utils/ZipFiles.java +++ b/file/src/main/java/org/apache/pulsar/io/file/utils/ZipFiles.java @@ -20,6 +20,7 @@ import java.io.BufferedInputStream; import java.io.BufferedReader; +import java.io.Closeable; import java.io.DataInputStream; import java.io.File; import java.io.FileInputStream; @@ -28,9 +29,12 @@ import java.io.UncheckedIOException; import java.nio.file.Files; import java.nio.file.Path; -import java.util.ArrayList; -import java.util.List; +import java.util.Iterator; +import java.util.NoSuchElementException; +import java.util.Spliterator; +import java.util.Spliterators; import java.util.stream.Stream; +import java.util.stream.StreamSupport; import java.util.zip.ZipEntry; import java.util.zip.ZipInputStream; @@ -53,38 +57,102 @@ public static boolean isZip(File f) { } /** - * Get a stream of lines from every file entry of a zip file, similar to + * Get a lazily loaded stream of lines from every file entry of a zip file, similar to * {@link Files#lines(java.nio.file.Path)}. * + *

A {@link ZipInputStream} yields no data until it is positioned onto an entry via + * {@link ZipInputStream#getNextEntry()}; without that call the reader sees an empty stream + * and no lines are produced. The returned stream lazily walks every file entry and emits + * their lines in entry order, so multi-entry archives contribute all of their lines without + * buffering the whole archive in memory. The caller must close the returned stream to + * release the underlying file. + * * @param path * The path to the zipped file. * @return stream with the lines of all file entries, in entry order. */ public static Stream lines(Path path) { - // A ZipInputStream returns no data until it is positioned onto an entry via - // getNextEntry(); without that call the reader sees an empty stream and no lines are - // produced. Read through every file entry so multi-entry archives contribute all of - // their lines, then return the collected lines as a stream (the archive must be fully - // read to walk its entries, so this cannot be lazy the way GZipFiles.lines is). - List lines = new ArrayList<>(); - try (ZipInputStream zipStream = new ZipInputStream(Files.newInputStream(path))) { - ZipEntry entry; - while ((entry = zipStream.getNextEntry()) != null) { - if (entry.isDirectory()) { - continue; - } - // Read the current entry to its end. read() returns -1 at the entry boundary, - // so the reader stops before the next entry; do not close it, as that would - // close the shared zipStream mid-iteration. - BufferedReader reader = new BufferedReader(new InputStreamReader(zipStream)); - String line; - while ((line = reader.readLine()) != null) { - lines.add(line); - } - } + ZipInputStream zipStream; + try { + zipStream = new ZipInputStream(Files.newInputStream(path)); } catch (IOException e) { throw new UncheckedIOException(e); } - return lines.stream(); + Spliterator spliterator = Spliterators.spliteratorUnknownSize( + new ZipLineIterator(zipStream), Spliterator.ORDERED | Spliterator.NONNULL); + return StreamSupport.stream(spliterator, false).onClose(() -> closeSafely(zipStream)); + } + + /** + * Iterates the lines of every non-directory entry of a zip stream in entry order, advancing + * to the next entry when the current one is exhausted. Entry boundaries are line boundaries: + * a fresh {@link BufferedReader} is used for each entry, so an entry whose content does not + * end with a newline does not merge its trailing text into the next entry's first line. + */ + private static final class ZipLineIterator implements Iterator { + private final ZipInputStream zipStream; + private BufferedReader reader; + private String nextLine; + + ZipLineIterator(ZipInputStream zipStream) { + this.zipStream = zipStream; + } + + @Override + public boolean hasNext() { + if (nextLine == null) { + nextLine = readNextLine(); + } + return nextLine != null; + } + + @Override + public String next() { + if (!hasNext()) { + throw new NoSuchElementException(); + } + String line = nextLine; + nextLine = null; + return line; + } + + private String readNextLine() { + try { + while (true) { + if (reader != null) { + String line = reader.readLine(); + if (line != null) { + return line; + } + // Current entry is exhausted; drop its reader but keep the shared + // zipStream open so we can position onto the next entry. + reader = null; + } + ZipEntry entry = zipStream.getNextEntry(); + if (entry == null) { + return null; + } + if (!entry.isDirectory()) { + // Wrap the shared zipStream in a fresh reader for this entry. read() + // returns -1 at the entry boundary, so the reader stops before the next + // entry; the reader is intentionally not closed, as that would close the + // shared zipStream mid-iteration. + reader = new BufferedReader(new InputStreamReader(zipStream)); + } + } + } catch (IOException e) { + throw new UncheckedIOException(e); + } + } + } + + private static void closeSafely(Closeable closeable) { + if (closeable != null) { + try { + closeable.close(); + } catch (IOException e) { + // Ignore + } + } } }