From 9f5efb1dbc36310c3fe4b03e337b2f1766652f8e Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Wed, 15 Jul 2026 06:45:34 -0700 Subject: [PATCH 1/3] [improve][test] Add File source integration test Drives FileSource through its real open() -> worker pool -> consume() path and asserts the emitted record bytes, rather than exercising the internal worker threads in isolation as the existing unit tests do. Covers plain-text line splitting, gzip decode, recursive traversal, file filtering, hidden-file exclusion, delete-after-process, processedFileSuffix rename, polling pickup of files created after open(), and the empty-directory path. All waits are deadline-bounded so an under-delivering pipeline fails fast instead of hanging CI. No external service required. Fixes #108 --- .../io/file/FileSourceIntegrationTest.java | 301 ++++++++++++++++++ 1 file changed, 301 insertions(+) create mode 100644 file/src/test/java/org/apache/pulsar/io/file/FileSourceIntegrationTest.java diff --git a/file/src/test/java/org/apache/pulsar/io/file/FileSourceIntegrationTest.java b/file/src/test/java/org/apache/pulsar/io/file/FileSourceIntegrationTest.java new file mode 100644 index 0000000000..ff78a8a2c1 --- /dev/null +++ b/file/src/test/java/org/apache/pulsar/io/file/FileSourceIntegrationTest.java @@ -0,0 +1,301 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.pulsar.io.file; + +import static org.testng.Assert.assertEquals; +import static org.testng.Assert.assertFalse; +import static org.testng.Assert.assertTrue; +import java.io.File; +import java.io.IOException; +import java.io.OutputStream; +import java.nio.charset.StandardCharsets; +import java.nio.file.FileVisitResult; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.SimpleFileVisitor; +import java.nio.file.attribute.BasicFileAttributes; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.zip.GZIPOutputStream; +import org.apache.pulsar.functions.api.Record; +import org.apache.pulsar.io.core.SourceContext; +import org.mockito.Mockito; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +/** + * End-to-end tests that drive the connector through its real public entry point + * ({@link FileSource#open}) and assert the records it emits, rather than exercising + * the internal worker threads in isolation as the unit tests do. + * + *

The source is a {@link org.apache.pulsar.io.core.PushSource}: its worker threads + * call {@code consume(record)}, which the framework drains via {@link FileSource#read()}. + * A background reader thread pulls those records into {@link #collected} so assertions can + * run against what the connector actually delivered. + * + *

No external service is required; every case is backed by a temp directory. A short + * polling interval keeps the tests fast, and every wait is deadline-bounded so an + * under-delivering pipeline fails fast instead of hanging CI. + */ +public class FileSourceIntegrationTest { + + private static final long POLLING_INTERVAL_MS = 300L; + private static final long DELIVERY_TIMEOUT_MS = 15_000L; + + private Path directory; + private FileSource source; + private Thread readerThread; + private List> collected; + + @BeforeMethod(alwaysRun = true) + public void init() throws IOException { + directory = Files.createTempDirectory("pulsar-io-file-it"); + } + + @AfterMethod(alwaysRun = true) + public void tearDown() throws Exception { + if (readerThread != null) { + readerThread.interrupt(); + readerThread.join(2000); + } + if (source != null) { + source.close(); + } + deleteRecursively(directory); + } + + // ------------------------------------------------------------------ tests + + @Test + public void testPlainTextFileEmitsOneRecordPerLine() throws Exception { + writeText("data.txt", "alpha", "bravo", "charlie"); + + startSource(baseConfig()); + assertTrue(awaitAtLeast(3, DELIVERY_TIMEOUT_MS), + "expected 3 records, got " + collected.size()); + + Map byKey = indexByKey(); + assertEquals(byKey.size(), 3); + assertEquals(byKey.get("data.txt_1"), "alpha"); + assertEquals(byKey.get("data.txt_2"), "bravo"); + assertEquals(byKey.get("data.txt_3"), "charlie"); + } + + @Test + public void testGzipFileIsDecoded() throws Exception { + writeGzip("data.txt.gz", "one", "two"); + + startSource(baseConfig()); + assertTrue(awaitAtLeast(2, DELIVERY_TIMEOUT_MS), + "expected 2 records, got " + collected.size()); + + Map byKey = indexByKey(); + assertEquals(byKey.get("data.txt.gz_1"), "one"); + assertEquals(byKey.get("data.txt.gz_2"), "two"); + } + + @Test + public void testRecursiveDirectoriesAreTraversed() throws Exception { + writeText("top.txt", "top"); + Path nested = Files.createDirectories(directory.resolve("sub/deeper")); + Files.write(nested.resolve("bottom.txt"), "bottom".getBytes(StandardCharsets.UTF_8)); + + Map config = baseConfig(); + config.put("recurse", true); + startSource(config); + + assertTrue(awaitAtLeast(2, DELIVERY_TIMEOUT_MS), + "expected 2 records across nested dirs, got " + collected.size()); + Map byKey = indexByKey(); + assertEquals(byKey.get("top.txt_1"), "top"); + assertEquals(byKey.get("bottom.txt_1"), "bottom"); + } + + @Test + public void testFileFilterExcludesNonMatchingFiles() throws Exception { + writeText("keep.csv", "kept"); + writeText("skip.log", "skipped"); + + Map config = baseConfig(); + config.put("fileFilter", ".*\\.csv"); + startSource(config); + + assertTrue(awaitAtLeast(1, DELIVERY_TIMEOUT_MS)); + // Give the listing loop a couple more cycles to (incorrectly) pick up the .log file. + assertFalse(awaitAtLeast(2, 3 * POLLING_INTERVAL_MS), + "only the .csv file should have been consumed"); + assertEquals(indexByKey().get("keep.csv_1"), "kept"); + } + + @Test + public void testHiddenFilesAreIgnored() throws Exception { + writeText("visible.txt", "seen"); + writeText(".hidden.txt", "unseen"); + + startSource(baseConfig()); + + assertTrue(awaitAtLeast(1, DELIVERY_TIMEOUT_MS)); + assertFalse(awaitAtLeast(2, 3 * POLLING_INTERVAL_MS), + "hidden files should not be consumed"); + assertEquals(indexByKey().get("visible.txt_1"), "seen"); + } + + @Test + public void testProcessedFileIsDeletedByDefault() throws Exception { + File file = writeText("ephemeral.txt", "gone"); + + startSource(baseConfig()); + assertTrue(awaitAtLeast(1, DELIVERY_TIMEOUT_MS)); + + assertTrue(awaitFileGone(file, DELIVERY_TIMEOUT_MS), + "default keepFile=false should delete the processed file"); + } + + @Test + public void testProcessedFileSuffixRenamesFile() throws Exception { + File file = writeText("record.txt", "kept-with-suffix"); + + Map config = baseConfig(); + config.put("processedFileSuffix", ".done"); + startSource(config); + + assertTrue(awaitAtLeast(1, DELIVERY_TIMEOUT_MS)); + File renamed = new File(file.getParentFile(), file.getName() + ".done"); + assertTrue(awaitFileGone(file, DELIVERY_TIMEOUT_MS), "original should be moved away"); + assertTrue(renamed.exists(), "processed file should be renamed with the configured suffix"); + } + + @Test + public void testFileAppearingAfterOpenIsPickedUpByPolling() throws Exception { + startSource(baseConfig()); + // Nothing on disk yet: the reader must stay empty until a file appears. + assertFalse(awaitAtLeast(1, 2 * POLLING_INTERVAL_MS)); + + writeText("late.txt", "arrived-late"); + assertTrue(awaitAtLeast(1, DELIVERY_TIMEOUT_MS), + "polling loop should pick up a file created after open()"); + assertEquals(indexByKey().get("late.txt_1"), "arrived-late"); + } + + @Test + public void testEmptyDirectoryEmitsNothing() throws Exception { + startSource(baseConfig()); + assertFalse(awaitAtLeast(1, 2 * POLLING_INTERVAL_MS), + "an empty input directory should yield no records"); + } + + // ---------------------------------------------------------------- helpers + + private Map baseConfig() { + Map config = new HashMap<>(); + config.put("inputDirectory", directory.toString()); + config.put("pollingInterval", POLLING_INTERVAL_MS); + config.put("numWorkers", 1); + return config; + } + + private void startSource(Map config) throws Exception { + source = new FileSource(); + source.open(config, Mockito.mock(SourceContext.class)); + collected = new CopyOnWriteArrayList<>(); + readerThread = new Thread(() -> { + try { + while (!Thread.currentThread().isInterrupted()) { + Record record = source.read(); + if (record != null) { + collected.add(record); + } + } + } catch (Exception e) { + // read() throws InterruptedException on teardown; terminate quietly. + } + }, "file-source-it-reader"); + readerThread.setDaemon(true); + readerThread.start(); + } + + /** Polls until at least {@code count} records have been collected or the deadline passes. */ + private boolean awaitAtLeast(int count, long timeoutMillis) throws InterruptedException { + long deadline = System.currentTimeMillis() + timeoutMillis; + while (System.currentTimeMillis() < deadline) { + if (collected.size() >= count) { + return true; + } + Thread.sleep(25); + } + return collected.size() >= count; + } + + private boolean awaitFileGone(File file, long timeoutMillis) throws InterruptedException { + long deadline = System.currentTimeMillis() + timeoutMillis; + while (System.currentTimeMillis() < deadline) { + if (!file.exists()) { + return true; + } + Thread.sleep(25); + } + return !file.exists(); + } + + /** Collapses collected records into key -> value(UTF-8); duplicate keys keep the last value. */ + private Map indexByKey() { + Map byKey = new LinkedHashMap<>(); + for (Record record : collected) { + byKey.put(record.getKey().orElse(null), new String(record.getValue(), StandardCharsets.UTF_8)); + } + return byKey; + } + + private File writeText(String name, String... lines) throws IOException { + Path path = directory.resolve(name); + Files.write(path, String.join("\n", lines).getBytes(StandardCharsets.UTF_8)); + return path.toFile(); + } + + private File writeGzip(String name, String... lines) throws IOException { + Path path = directory.resolve(name); + try (OutputStream out = new GZIPOutputStream(Files.newOutputStream(path))) { + out.write(String.join("\n", lines).getBytes(StandardCharsets.UTF_8)); + } + return path.toFile(); + } + + private static void deleteRecursively(Path root) throws IOException { + if (root == null || !Files.exists(root)) { + return; + } + Files.walkFileTree(root, new SimpleFileVisitor() { + @Override + public FileVisitResult visitFile(Path file, BasicFileAttributes attrs) throws IOException { + Files.deleteIfExists(file); + return FileVisitResult.CONTINUE; + } + + @Override + public FileVisitResult postVisitDirectory(Path dir, IOException exc) throws IOException { + Files.deleteIfExists(dir); + return FileVisitResult.CONTINUE; + } + }); + } +} From 2712803fc7624732f8d0658138646ec898bf26f2 Mon Sep 17 00:00:00 2001 From: David Kjerrumgaard <35466513+david-streamlio@users.noreply.github.com> Date: Wed, 15 Jul 2026 07:31:12 -0700 Subject: [PATCH 2/3] Potential fix for pull request finding Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com> --- .../org/apache/pulsar/io/file/FileSourceIntegrationTest.java | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/file/src/test/java/org/apache/pulsar/io/file/FileSourceIntegrationTest.java b/file/src/test/java/org/apache/pulsar/io/file/FileSourceIntegrationTest.java index ff78a8a2c1..363006a3ba 100644 --- a/file/src/test/java/org/apache/pulsar/io/file/FileSourceIntegrationTest.java +++ b/file/src/test/java/org/apache/pulsar/io/file/FileSourceIntegrationTest.java @@ -226,8 +226,10 @@ private void startSource(Map config) throws Exception { collected.add(record); } } + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); } catch (Exception e) { - // read() throws InterruptedException on teardown; terminate quietly. + throw new RuntimeException("Unexpected exception from FileSource.read()", e); } }, "file-source-it-reader"); readerThread.setDaemon(true); From a8b2c8251e18299321a3df2244ca142632a94803 Mon Sep 17 00:00:00 2001 From: david-streamlio <35466513+david-streamlio@users.noreply.github.com> Date: Wed, 15 Jul 2026 07:59:44 -0700 Subject: [PATCH 3/3] [improve][test] Address review comments on File source integration test - Write all test input files with Charset.defaultCharset() to match the connector (FileConsumerThread reads via Files.lines(defaultCharset) and re-encodes with line.getBytes()), avoiding platform-dependent failures. - Decode emitted record values with Charset.defaultCharset() for the same reason. - testHiddenFilesAreIgnored: set fileFilter to '.*' so the hidden file is excluded solely by ignoreHiddenFiles, actually validating that behavior. - tearDown: interrupt the reader, close the source to stop the worker pool and unblock read(), then join and assert the reader stopped; keep cleanup in a finally so a failing close() cannot leak the temp directory. - Remove now-unused StandardCharsets import. --- .../io/file/FileSourceIntegrationTest.java | 41 +++++++++++++------ 1 file changed, 28 insertions(+), 13 deletions(-) diff --git a/file/src/test/java/org/apache/pulsar/io/file/FileSourceIntegrationTest.java b/file/src/test/java/org/apache/pulsar/io/file/FileSourceIntegrationTest.java index 363006a3ba..51606e2282 100644 --- a/file/src/test/java/org/apache/pulsar/io/file/FileSourceIntegrationTest.java +++ b/file/src/test/java/org/apache/pulsar/io/file/FileSourceIntegrationTest.java @@ -24,7 +24,7 @@ import java.io.File; import java.io.IOException; import java.io.OutputStream; -import java.nio.charset.StandardCharsets; +import java.nio.charset.Charset; import java.nio.file.FileVisitResult; import java.nio.file.Files; import java.nio.file.Path; @@ -74,14 +74,25 @@ public void init() throws IOException { @AfterMethod(alwaysRun = true) public void tearDown() throws Exception { - if (readerThread != null) { - readerThread.interrupt(); - readerThread.join(2000); + // Interrupt the reader first, then close the source so its worker pool stops and any + // blocking read() is unblocked, then join the reader before asserting it terminated. + // Cleanup runs in a finally so a failing close() never leaks the temp directory. + try { + if (readerThread != null) { + readerThread.interrupt(); + } + if (source != null) { + source.close(); + } + if (readerThread != null) { + readerThread.join(2000); + } + } finally { + deleteRecursively(directory); } - if (source != null) { - source.close(); + if (readerThread != null) { + assertFalse(readerThread.isAlive(), "reader thread should have stopped after teardown"); } - deleteRecursively(directory); } // ------------------------------------------------------------------ tests @@ -118,7 +129,7 @@ public void testGzipFileIsDecoded() throws Exception { public void testRecursiveDirectoriesAreTraversed() throws Exception { writeText("top.txt", "top"); Path nested = Files.createDirectories(directory.resolve("sub/deeper")); - Files.write(nested.resolve("bottom.txt"), "bottom".getBytes(StandardCharsets.UTF_8)); + Files.write(nested.resolve("bottom.txt"), "bottom".getBytes(Charset.defaultCharset())); Map config = baseConfig(); config.put("recurse", true); @@ -152,7 +163,11 @@ public void testHiddenFilesAreIgnored() throws Exception { writeText("visible.txt", "seen"); writeText(".hidden.txt", "unseen"); - startSource(baseConfig()); + Map config = baseConfig(); + // Use a filter that DOES match dotfiles so the hidden file is excluded solely by + // ignoreHiddenFiles (which defaults to true), not by the default fileFilter ([^.].*). + config.put("fileFilter", ".*"); + startSource(config); assertTrue(awaitAtLeast(1, DELIVERY_TIMEOUT_MS)); assertFalse(awaitAtLeast(2, 3 * POLLING_INTERVAL_MS), @@ -259,25 +274,25 @@ private boolean awaitFileGone(File file, long timeoutMillis) throws InterruptedE return !file.exists(); } - /** Collapses collected records into key -> value(UTF-8); duplicate keys keep the last value. */ + /** Collapses collected records into key -> value(default charset); duplicate keys keep the last value. */ private Map indexByKey() { Map byKey = new LinkedHashMap<>(); for (Record record : collected) { - byKey.put(record.getKey().orElse(null), new String(record.getValue(), StandardCharsets.UTF_8)); + byKey.put(record.getKey().orElse(null), new String(record.getValue(), Charset.defaultCharset())); } return byKey; } private File writeText(String name, String... lines) throws IOException { Path path = directory.resolve(name); - Files.write(path, String.join("\n", lines).getBytes(StandardCharsets.UTF_8)); + Files.write(path, String.join("\n", lines).getBytes(Charset.defaultCharset())); return path.toFile(); } private File writeGzip(String name, String... lines) throws IOException { Path path = directory.resolve(name); try (OutputStream out = new GZIPOutputStream(Files.newOutputStream(path))) { - out.write(String.join("\n", lines).getBytes(StandardCharsets.UTF_8)); + out.write(String.join("\n", lines).getBytes(Charset.defaultCharset())); } return path.toFile(); }