You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "david-streamlio (via GitHub)" <gi...@apache.org> on 2026/07/15 14:02:43 UTC
[PR] [fix][io] File source: read zip entries so zip files emit records [pulsar-connectors]
david-streamlio opened a new pull request, #117:
URL: https://github.com/apache/pulsar-connectors/pull/117
Fixes #114
### Motivation
The File source silently emits **zero records** for zip-formatted input files, despite `FileSource` advertising zip support. `ZipFiles.lines()` opened a `ZipInputStream` and wrapped it in a `BufferedReader`, but never called `getNextEntry()`. A `ZipInputStream` returns EOF from `read()` until it is positioned onto an entry, so the reader saw an empty stream — and the file was then deleted/renamed as if successfully processed.
The existing `ZipFilesTest.streamZipFileTest` did not catch this: it asserted inside a `forEachOrdered` that ran zero times and swallowed exceptions, so it passed vacuously.
### Modifications
- `ZipFiles.lines()` now iterates every file entry (`getNextEntry()`), reads each entry to its end, and returns the collected lines in entry order. Single- and multi-entry archives both deliver their content. (Walking a zip's entries requires reading the archive, so the result is eager rather than lazy — unlike the single-member `GZipFiles.lines`.)
- `ZipFilesTest.streamZipFileTest` is now non-vacuous: it asserts the exact nine lines of the fixture (`Line 1`..`Line 9`) and no longer swallows exceptions.
- New `streamMultiEntryZipFileTest` builds a two-entry archive in a temp file and asserts all lines from both entries are returned, in order.
### Verifying this change
```
$ ./gradlew :file:check
BUILD SUCCESSFUL
```
Before the fix, `ZipFiles.lines()` returned an empty stream (verified against the JDK `ZipInputStream`); after it, the fixture yields its nine lines and the multi-entry archive yields all five.
### Documentation
- [x] `doc-not-needed` (bug fix + tests)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] File source: read zip entries so zip files emit records [pulsar-connectors]
Posted by "david-streamlio (via GitHub)" <gi...@apache.org>.
david-streamlio merged PR #117:
URL: https://github.com/apache/pulsar-connectors/pull/117
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] File source: read zip entries so zip files emit records [pulsar-connectors]
Posted by "david-streamlio (via GitHub)" <gi...@apache.org>.
david-streamlio commented on code in PR #117:
URL: https://github.com/apache/pulsar-connectors/pull/117#discussion_r3588229596
##########
file/src/main/java/org/apache/pulsar/io/file/utils/ZipFiles.java:
##########
@@ -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<String> 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).
Review Comment:
Good catch on both points, addressed in 1e0730f.
`ZipFiles.lines()` no longer buffers into a `List`. It now returns a lazy stream backed by a line iterator that walks the entries one at a time (opening a fresh reader per file entry so entry boundaries stay line boundaries), and closes the `ZipInputStream` via `onClose` when the returned stream is closed. `FileConsumerThread` already consumes it inside a try-with-resources, so the stream is closed and per-line back-pressure is preserved. This matches the lazy contract of `GZipFiles.lines`. The inaccurate "cannot be lazy" comment is gone.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [fix][io] File source: read zip entries so zip files emit records [pulsar-connectors]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #117:
URL: https://github.com/apache/pulsar-connectors/pull/117#discussion_r3587937045
##########
file/src/main/java/org/apache/pulsar/io/file/utils/ZipFiles.java:
##########
@@ -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<String> 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).
Review Comment:
`ZipFiles.lines()` now reads the entire zip into a `List<String>` before returning a stream. This changes behavior from the previous lazy stream and can significantly increase memory usage and delay processing for large archives (notably `FileConsumerThread` consumes the returned `Stream` line-by-line). Also, the comment stating the result "cannot be lazy" is inaccurate: you can iterate entries and yield lines lazily while keeping the `ZipInputStream` open and closing it when the returned stream is closed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org