This is an automated email from the ASF dual-hosted git repository.
david-streamlio pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar-connectors.git
The following commit(s) were added to refs/heads/master by this push:
new eb244a72 [fix][io] Fix skipped file listing loop runs in file
connector (#14)
eb244a72 is described below
commit eb244a726a9a201d6db4052520d6f35458051f2d
Author: Philipp Dolif <[email protected]>
AuthorDate: Fri Jul 10 01:16:30 2026 +0200
[fix][io] Fix skipped file listing loop runs in file connector (#14)
* [fix][io] Fix skipped file listing loop runs in file connector
* FileListingThread: Change from Thread to Runnable
* Rename FileListingThread to FileListingTask
* Respect interrupts in FileListingTask polling loop
---------
Co-authored-by: David Kjerrumgaard
<[email protected]>
---
...FileListingThread.java => FileListingTask.java} | 22 ++++---
.../java/org/apache/pulsar/io/file/FileSource.java | 4 +-
.../apache/pulsar/io/file/AbstractFileTest.java | 2 +-
.../pulsar/io/file/FileConsumerThreadTest.java | 12 ++--
...ingThreadTest.java => FileListingTaskTest.java} | 75 +++++++++++++++-------
.../pulsar/io/file/ProcessedFileThreadTest.java | 24 +++----
6 files changed, 85 insertions(+), 54 deletions(-)
diff --git
a/file/src/main/java/org/apache/pulsar/io/file/FileListingThread.java
b/file/src/main/java/org/apache/pulsar/io/file/FileListingTask.java
similarity index 93%
rename from file/src/main/java/org/apache/pulsar/io/file/FileListingThread.java
rename to file/src/main/java/org/apache/pulsar/io/file/FileListingTask.java
index 10a3c862..e9e3dec9 100644
--- a/file/src/main/java/org/apache/pulsar/io/file/FileListingThread.java
+++ b/file/src/main/java/org/apache/pulsar/io/file/FileListingTask.java
@@ -35,11 +35,11 @@ import java.util.regex.Pattern;
import org.apache.commons.lang3.StringUtils;
/**
- * Worker thread that checks the configured input directory for
+ * Worker that checks the configured input directory for
* files that meet the provided filtering criteria, and publishes
* them to a work queue for processing by the FileConsumerThreads.
*/
-public class FileListingThread extends Thread {
+public class FileListingTask implements Runnable {
private final AtomicLong queueLastUpdated = new AtomicLong(0L);
private final Lock listingLock = new ReentrantLock();
@@ -62,10 +62,10 @@ public class FileListingThread extends Thread {
private final boolean keepOriginal;
private final long pollingInterval;
- public FileListingThread(FileSourceConfig fileConfig,
- BlockingQueue<File> workQueue,
- BlockingQueue<File> inProcess,
- BlockingQueue<File> recentlyProcessed) {
+ public FileListingTask(FileSourceConfig fileConfig,
+ BlockingQueue<File> workQueue,
+ BlockingQueue<File> inProcess,
+ BlockingQueue<File> recentlyProcessed) {
this.workQueue = workQueue;
this.inProcess = inProcess;
this.recentlyProcessed = recentlyProcessed;
@@ -77,9 +77,10 @@ public class FileListingThread extends Thread {
fileFilterRef.set(createFileFilter(fileConfig));
}
+ @Override
public void run() {
- while (true) {
- if ((queueLastUpdated.get() < System.currentTimeMillis() -
pollingInterval) && listingLock.tryLock()) {
+ while (!Thread.currentThread().isInterrupted()) {
+ if ((queueLastUpdated.get() <= System.currentTimeMillis() -
pollingInterval) && listingLock.tryLock()) {
try {
// Prune tracked files that are gone from disk (processed
and then renamed or
// deleted). This must happen before the listing snapshot:
a file that
@@ -123,9 +124,10 @@ public class FileListingThread extends Thread {
}
try {
- sleep(pollingInterval - 1);
+ Thread.sleep(pollingInterval);
} catch (InterruptedException e) {
- // Just ignore
+ Thread.currentThread().interrupt();
+ break;
}
}
}
diff --git a/file/src/main/java/org/apache/pulsar/io/file/FileSource.java
b/file/src/main/java/org/apache/pulsar/io/file/FileSource.java
index c201977d..85d46c51 100644
--- a/file/src/main/java/org/apache/pulsar/io/file/FileSource.java
+++ b/file/src/main/java/org/apache/pulsar/io/file/FileSource.java
@@ -45,9 +45,9 @@ public class FileSource extends PushSource<byte[]> {
FileSourceConfig fileConfig = FileSourceConfig.load(config);
fileConfig.validate();
- // One extra for the File listing thread, and another for the cleanup
thread
+ // One extra for the File listing task, and another for the cleanup
thread
executor = Executors.newFixedThreadPool(fileConfig.getNumWorkers() +
2);
- executor.execute(new FileListingThread(fileConfig, workQueue,
inProcess, recentlyProcessed));
+ executor.execute(new FileListingTask(fileConfig, workQueue, inProcess,
recentlyProcessed));
executor.execute(new ProcessedFileThread(fileConfig,
recentlyProcessed));
for (int idx = 0; idx < fileConfig.getNumWorkers(); idx++) {
diff --git a/file/src/test/java/org/apache/pulsar/io/file/AbstractFileTest.java
b/file/src/test/java/org/apache/pulsar/io/file/AbstractFileTest.java
index 1cfa2202..70e820a3 100644
--- a/file/src/test/java/org/apache/pulsar/io/file/AbstractFileTest.java
+++ b/file/src/test/java/org/apache/pulsar/io/file/AbstractFileTest.java
@@ -49,7 +49,7 @@ public abstract class AbstractFileTest {
protected BlockingQueue<File> producedFiles;
protected TestFileGenerator generatorThread;
- protected FileListingThread listingThread;
+ protected FileListingTask listingTask;
protected ExecutorService executor;
protected Path directory;
diff --git
a/file/src/test/java/org/apache/pulsar/io/file/FileConsumerThreadTest.java
b/file/src/test/java/org/apache/pulsar/io/file/FileConsumerThreadTest.java
index 6f1d958d..6f2a182e 100644
--- a/file/src/test/java/org/apache/pulsar/io/file/FileConsumerThreadTest.java
+++ b/file/src/test/java/org/apache/pulsar/io/file/FileConsumerThreadTest.java
@@ -50,9 +50,9 @@ public class FileConsumerThreadTest extends AbstractFileTest {
try {
generateFiles(1);
- listingThread = new FileListingThread(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
+ listingTask = new FileListingTask(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
consumerThread = new FileConsumerThread(consumer, workQueue,
inProcess, recentlyProcessed);
- executor.execute(listingThread);
+ executor.execute(listingTask);
executor.execute(consumerThread);
Thread.sleep(2000);
@@ -85,9 +85,9 @@ public class FileConsumerThreadTest extends AbstractFileTest {
try {
generateFiles(50, 2);
- listingThread = new FileListingThread(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
+ listingTask = new FileListingTask(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
consumerThread = new FileConsumerThread(consumer, workQueue,
inProcess, recentlyProcessed);
- executor.execute(listingThread);
+ executor.execute(listingTask);
executor.execute(consumerThread);
Thread.sleep(2000);
@@ -120,9 +120,9 @@ public class FileConsumerThreadTest extends
AbstractFileTest {
try {
generateFiles(1, 10);
- listingThread = new FileListingThread(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
+ listingTask = new FileListingTask(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
consumerThread = new FileConsumerThread(consumer, workQueue,
inProcess, recentlyProcessed);
- executor.execute(listingThread);
+ executor.execute(listingTask);
executor.execute(consumerThread);
Thread.sleep(2000);
diff --git
a/file/src/test/java/org/apache/pulsar/io/file/FileListingThreadTest.java
b/file/src/test/java/org/apache/pulsar/io/file/FileListingTaskTest.java
similarity index 76%
rename from
file/src/test/java/org/apache/pulsar/io/file/FileListingThreadTest.java
rename to file/src/test/java/org/apache/pulsar/io/file/FileListingTaskTest.java
index 8df748d8..d4fdeef9 100644
--- a/file/src/test/java/org/apache/pulsar/io/file/FileListingThreadTest.java
+++ b/file/src/test/java/org/apache/pulsar/io/file/FileListingTaskTest.java
@@ -30,7 +30,7 @@ import java.util.concurrent.ExecutionException;
import org.testng.annotations.Test;
-public class FileListingThreadTest extends AbstractFileTest {
+public class FileListingTaskTest extends AbstractFileTest {
@Test
public final void singleFileTest() throws IOException {
@@ -40,8 +40,8 @@ public class FileListingThreadTest extends AbstractFileTest {
try {
generateFiles(1);
- listingThread = new FileListingThread(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
- executor.execute(listingThread);
+ listingTask = new FileListingTask(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
+ executor.execute(listingTask);
Thread.sleep(2000);
verify(producedFiles, times(1)).put(any(File.class));
verify(workQueue, times(1)).offer(any(File.class));
@@ -63,8 +63,8 @@ public class FileListingThreadTest extends AbstractFileTest {
try {
generateFiles(50);
- listingThread = new FileListingThread(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
- executor.execute(listingThread);
+ listingTask = new FileListingTask(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
+ executor.execute(listingTask);
Thread.sleep(2000);
verify(workQueue, times(50)).offer(any(File.class));
@@ -86,8 +86,8 @@ public class FileListingThreadTest extends AbstractFileTest {
try {
// Create 50 zero size files
generateFiles(50, 0);
- listingThread = new FileListingThread(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
- executor.execute(listingThread);
+ listingTask = new FileListingTask(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
+ executor.execute(listingTask);
Thread.sleep(2000);
verify(workQueue, times(0)).offer(any(File.class));
} catch (InterruptedException | ExecutionException e) {
@@ -106,8 +106,8 @@ public class FileListingThreadTest extends AbstractFileTest
{
// Create 5 files that exceed the limit and 45 that don't
generateFiles(5, 1000);
generateFiles(45, 10);
- listingThread = new FileListingThread(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
- executor.execute(listingThread);
+ listingTask = new FileListingTask(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
+ executor.execute(listingTask);
Thread.sleep(2000);
verify(workQueue, times(45)).offer(any(File.class));
} catch (InterruptedException | ExecutionException e) {
@@ -127,8 +127,8 @@ public class FileListingThreadTest extends AbstractFileTest
{
try {
// Create 5 files that will be too "new" for processing
generateFiles(5);
- listingThread = new FileListingThread(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
- executor.execute(listingThread);
+ listingTask = new FileListingTask(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
+ executor.execute(listingTask);
Thread.sleep(2000);
verify(workQueue, times(0)).offer(any(File.class));
} catch (InterruptedException | ExecutionException e) {
@@ -152,8 +152,8 @@ public class FileListingThreadTest extends AbstractFileTest
{
// Create 5 files that will be too "old" for processing
generateFiles(5);
- listingThread = new FileListingThread(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
- executor.execute(listingThread);
+ listingTask = new FileListingTask(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
+ executor.execute(listingTask);
Thread.sleep(2000);
verify(workQueue, times(5)).offer(any(File.class));
} catch (InterruptedException | ExecutionException e) {
@@ -163,6 +163,35 @@ public class FileListingThreadTest extends
AbstractFileTest {
}
}
+ @Test
+ public void pollingIntervalTest() throws IOException {
+ int pollingInterval = 100;
+ int tolerance = 20;
+
+ Map<String, Object> map = new HashMap<>();
+ map.put("inputDirectory", directory.toString());
+ map.put("pollingInterval", pollingInterval);
+
+ try {
+ listingTask = new FileListingTask(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
+ executor.execute(listingTask);
+
+ generateFiles(1);
+ Thread.sleep(pollingInterval + tolerance);
+
+ verify(workQueue, times(1)).offer(any(File.class));
+
+ generateFiles(1);
+ Thread.sleep(pollingInterval + tolerance);
+
+ verify(workQueue, times(2)).offer(any(File.class));
+ } catch (InterruptedException | ExecutionException e) {
+ fail("Unable to generate files" + e.getLocalizedMessage());
+ } finally {
+ cleanUp();
+ }
+ }
+
@Test
public final void doRecurseTest() throws IOException {
@@ -176,8 +205,8 @@ public class FileListingThreadTest extends AbstractFileTest
{
// Create 5 files in a sub-folder
generateFiles(5, 1, directory.toString() + File.separator +
"sub-dir");
- listingThread = new FileListingThread(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
- executor.execute(listingThread);
+ listingTask = new FileListingTask(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
+ executor.execute(listingTask);
Thread.sleep(2000);
verify(workQueue, times(10)).offer(any(File.class));
} catch (InterruptedException | ExecutionException e) {
@@ -200,8 +229,8 @@ public class FileListingThreadTest extends AbstractFileTest
{
// Create 5 files in a sub-folder
generateFiles(5, 1, directory.toString() + File.separator +
"sub-dir");
- listingThread = new FileListingThread(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
- executor.execute(listingThread);
+ listingTask = new FileListingTask(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
+ executor.execute(listingTask);
Thread.sleep(2000);
verify(workQueue, times(5)).offer(any(File.class));
} catch (InterruptedException | ExecutionException e) {
@@ -223,8 +252,8 @@ public class FileListingThreadTest extends AbstractFileTest
{
// Create 5 files in a sub-folder
generateFiles(5, 1, directory.toString() + File.separator +
"sub-dir-a");
generateFiles(5, 1, directory.toString() + File.separator +
"dir-b");
- listingThread = new FileListingThread(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
- executor.execute(listingThread);
+ listingTask = new FileListingTask(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
+ executor.execute(listingTask);
Thread.sleep(2000);
verify(workQueue, times(5)).offer(any(File.class));
} catch (InterruptedException | ExecutionException e) {
@@ -246,8 +275,8 @@ public class FileListingThreadTest extends AbstractFileTest
{
try {
generateFiles(5, 1, directory.toString(), ".txt");
generateFiles(1, 1, directory.toString(), processedFileSuffix);
- listingThread = new FileListingThread(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
- executor.execute(listingThread);
+ listingTask = new FileListingTask(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
+ executor.execute(listingTask);
Thread.sleep(2000);
verify(workQueue, times(5)).offer(any(File.class));
} catch (InterruptedException | ExecutionException e) {
@@ -269,8 +298,8 @@ public class FileListingThreadTest extends AbstractFileTest
{
try {
generateFiles(5, 1, directory.toString(), ".txt");
generateFiles(1, 1, directory.toString(), processedFileSuffix);
- listingThread = new FileListingThread(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
- executor.execute(listingThread);
+ listingTask = new FileListingTask(FileSourceConfig.load(map),
workQueue, inProcess, recentlyProcessed);
+ executor.execute(listingTask);
Thread.sleep(2000);
verify(workQueue, times(6)).offer(any(File.class));
} catch (InterruptedException | ExecutionException e) {
diff --git
a/file/src/test/java/org/apache/pulsar/io/file/ProcessedFileThreadTest.java
b/file/src/test/java/org/apache/pulsar/io/file/ProcessedFileThreadTest.java
index 6e4f015d..4d109cb7 100644
--- a/file/src/test/java/org/apache/pulsar/io/file/ProcessedFileThreadTest.java
+++ b/file/src/test/java/org/apache/pulsar/io/file/ProcessedFileThreadTest.java
@@ -80,10 +80,10 @@ public class ProcessedFileThreadTest extends
AbstractFileTest {
try {
generateFiles(1);
fileConfig = FileSourceConfig.load(map);
- listingThread = new FileListingThread(fileConfig, workQueue,
inProcess, recentlyProcessed);
+ listingTask = new FileListingTask(fileConfig, workQueue,
inProcess, recentlyProcessed);
consumerThread = new FileConsumerThread(consumer, workQueue,
inProcess, recentlyProcessed);
cleanupThread = new ProcessedFileThread(fileConfig,
recentlyProcessed);
- executor.execute(listingThread);
+ executor.execute(listingTask);
executor.execute(consumerThread);
executor.execute(cleanupThread);
Thread.sleep(2000);
@@ -119,10 +119,10 @@ public class ProcessedFileThreadTest extends
AbstractFileTest {
try {
generateFiles(50);
fileConfig = FileSourceConfig.load(map);
- listingThread = new FileListingThread(fileConfig, workQueue,
inProcess, recentlyProcessed);
+ listingTask = new FileListingTask(fileConfig, workQueue,
inProcess, recentlyProcessed);
consumerThread = new FileConsumerThread(consumer, workQueue,
inProcess, recentlyProcessed);
cleanupThread = new ProcessedFileThread(fileConfig,
recentlyProcessed);
- executor.execute(listingThread);
+ executor.execute(listingTask);
executor.execute(consumerThread);
executor.execute(cleanupThread);
Thread.sleep(2000);
@@ -160,10 +160,10 @@ public class ProcessedFileThreadTest extends
AbstractFileTest {
try {
generateFiles(1);
fileConfig = FileSourceConfig.load(map);
- listingThread = new FileListingThread(fileConfig, workQueue,
inProcess, recentlyProcessed);
+ listingTask = new FileListingTask(fileConfig, workQueue,
inProcess, recentlyProcessed);
consumerThread = new FileConsumerThread(consumer, workQueue,
inProcess, recentlyProcessed);
cleanupThread = new ProcessedFileThread(fileConfig,
recentlyProcessed);
- executor.execute(listingThread);
+ executor.execute(listingTask);
executor.execute(consumerThread);
executor.execute(cleanupThread);
Thread.sleep(7900); // Should pull the same file 5 times?
@@ -199,10 +199,10 @@ public class ProcessedFileThreadTest extends
AbstractFileTest {
directory.toString(), "continuous", ".txt",
getPermissions());
executor.execute(generatorThread);
- listingThread = new FileListingThread(fileConfig, workQueue,
inProcess, recentlyProcessed);
+ listingTask = new FileListingTask(fileConfig, workQueue,
inProcess, recentlyProcessed);
consumerThread = new FileConsumerThread(consumer, workQueue,
inProcess, recentlyProcessed);
cleanupThread = new ProcessedFileThread(fileConfig,
recentlyProcessed);
- executor.execute(listingThread);
+ executor.execute(listingTask);
executor.execute(consumerThread);
executor.execute(cleanupThread);
@@ -246,14 +246,14 @@ public class ProcessedFileThreadTest extends
AbstractFileTest {
directory.toString(), "continuous", ".txt",
getPermissions());
executor.execute(generatorThread);
- listingThread = new FileListingThread(fileConfig, workQueue,
inProcess, recentlyProcessed);
+ listingTask = new FileListingTask(fileConfig, workQueue,
inProcess, recentlyProcessed);
consumerThread = new FileConsumerThread(consumer, workQueue,
inProcess, recentlyProcessed);
FileConsumerThread consumerThread2 = new
FileConsumerThread(consumer, workQueue, inProcess,
recentlyProcessed);
FileConsumerThread consumerThread3 = new
FileConsumerThread(consumer, workQueue, inProcess,
recentlyProcessed);
cleanupThread = new ProcessedFileThread(fileConfig,
recentlyProcessed);
- executor.execute(listingThread);
+ executor.execute(listingTask);
executor.execute(consumerThread);
executor.execute(consumerThread2);
executor.execute(consumerThread3);
@@ -302,10 +302,10 @@ public class ProcessedFileThreadTest extends
AbstractFileTest {
"continuous", ".txt", getPermissions());
executor.execute(generatorThread);
- listingThread = new FileListingThread(fileConfig, workQueue,
inProcess, recentlyProcessed);
+ listingTask = new FileListingTask(fileConfig, workQueue,
inProcess, recentlyProcessed);
consumerThread = new FileConsumerThread(consumer, workQueue,
inProcess, recentlyProcessed);
cleanupThread = new ProcessedFileThread(fileConfig,
recentlyProcessed);
- executor.execute(listingThread);
+ executor.execute(listingTask);
executor.execute(consumerThread);
executor.execute(cleanupThread);
