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 10f9daee [fix][io] Prevent OOM in FileSource by bounding internal
queues (#82)
10f9daee is described below
commit 10f9daee19f86a66dd78a75c694cdc98e953060c
Author: David Kjerrumgaard <[email protected]>
AuthorDate: Fri Jul 10 06:21:35 2026 -0700
[fix][io] Prevent OOM in FileSource by bounding internal queues (#82)
* [fix][io] Prevent OOM in FileSource by bounding internal queues
* revert with old changes for specific getfile method
* [fix][io] Block instead of throwing when bounded queues are full
Address the review findings on the bounded-queue change:
FileConsumerThread produced into inProcess and recentlyProcessed with
BlockingQueue.add(), which throws IllegalStateException on a bounded
queue that is full rather than returning false — so the surrounding
do/while(!added) retry loop never ran. Once the queues are bounded, a
cleanup thread slower than the workers would kill the worker outright
and stall processing. Use put() so the producer blocks and applies
back-pressure. Existing tests verified add() on the spied queues and
follow the change.
FileSourceConfig.validate() permitted a null maxQueueSize, which
FileSource.open() then unboxed to size the queues, turning a config
error into a NullPointerException. Reject null explicitly.
FileSourceConfigTest hard-coded /tmp as a dummy input directory; use
the existing INPUT_DIRECTORY constant so the tests do not depend on
filesystem layout.
Add FileConsumerThreadBoundedQueueTest, which fills recentlyProcessed
and asserts the consumer blocks and survives; it fails against add()
with 'consumer thread died: expected 3 files, got 1'.
---------
Co-authored-by: Praveenkumar76 <[email protected]>
---
.../apache/pulsar/io/file/FileConsumerThread.java | 15 ++--
.../java/org/apache/pulsar/io/file/FileSource.java | 12 ++-
.../apache/pulsar/io/file/FileSourceConfig.java | 12 +++
.../file/FileConsumerThreadBoundedQueueTest.java | 90 ++++++++++++++++++++++
.../pulsar/io/file/FileConsumerThreadTest.java | 24 +++---
.../pulsar/io/file/FileSourceConfigTest.java | 65 ++++++++++++++++
.../pulsar/io/file/ProcessedFileThreadTest.java | 34 ++++----
7 files changed, 211 insertions(+), 41 deletions(-)
diff --git
a/file/src/main/java/org/apache/pulsar/io/file/FileConsumerThread.java
b/file/src/main/java/org/apache/pulsar/io/file/FileConsumerThread.java
index 7e0c9344..8845da49 100644
--- a/file/src/main/java/org/apache/pulsar/io/file/FileConsumerThread.java
+++ b/file/src/main/java/org/apache/pulsar/io/file/FileConsumerThread.java
@@ -56,10 +56,9 @@ public class FileConsumerThread extends Thread {
while (true) {
File file = workQueue.take();
- boolean added = false;
- do {
- added = inProcess.add(file);
- } while (!added);
+ // put(), not add(): add() throws IllegalStateException on a
bounded queue that is
+ // full, which would kill this worker. Blocking applies
back-pressure instead.
+ inProcess.put(file);
consumeFile(file);
}
@@ -68,7 +67,7 @@ public class FileConsumerThread extends Thread {
}
}
- private void consumeFile(File file) {
+ private void consumeFile(File file) throws InterruptedException {
final AtomicInteger idx = new AtomicInteger(1);
try (Stream<String> lines = getLines(file)) {
lines.forEachOrdered(line -> process(file, idx.getAndIncrement(),
line));
@@ -81,10 +80,8 @@ public class FileConsumerThread extends Thread {
removed = inProcess.remove(file);
} while (!removed);
- boolean added = false;
- do {
- added = recentlyProcessed.add(file);
- } while (!added);
+ // put(), not add(): see the comment in run().
+ recentlyProcessed.put(file);
}
}
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 85d46c51..70780929 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
@@ -36,15 +36,21 @@ import org.apache.pulsar.io.core.SourceContext;
public class FileSource extends PushSource<byte[]> {
private ExecutorService executor;
- private final BlockingQueue<File> workQueue = new LinkedBlockingQueue<>();
- private final BlockingQueue<File> inProcess = new LinkedBlockingQueue<>();
- private final BlockingQueue<File> recentlyProcessed = new
LinkedBlockingQueue<>();
+ private BlockingQueue<File> workQueue;
+ private BlockingQueue<File> inProcess;
+ private BlockingQueue<File> recentlyProcessed;
@Override
public void open(Map<String, Object> config, SourceContext sourceContext)
throws Exception {
FileSourceConfig fileConfig = FileSourceConfig.load(config);
fileConfig.validate();
+ // Initialize bounded queues based on configuration to prevent OOM
+ int queueCapacity = fileConfig.getMaxQueueSize();
+ workQueue = new LinkedBlockingQueue<>(queueCapacity);
+ inProcess = new LinkedBlockingQueue<>(queueCapacity);
+ recentlyProcessed = new LinkedBlockingQueue<>(queueCapacity);
+
// One extra for the File listing task, and another for the cleanup
thread
executor = Executors.newFixedThreadPool(fileConfig.getNumWorkers() +
2);
executor.execute(new FileListingTask(fileConfig, workQueue, inProcess,
recentlyProcessed));
diff --git a/file/src/main/java/org/apache/pulsar/io/file/FileSourceConfig.java
b/file/src/main/java/org/apache/pulsar/io/file/FileSourceConfig.java
index d9b1e1f3..40e11201 100644
--- a/file/src/main/java/org/apache/pulsar/io/file/FileSourceConfig.java
+++ b/file/src/main/java/org/apache/pulsar/io/file/FileSourceConfig.java
@@ -110,6 +110,12 @@ public class FileSourceConfig implements Serializable {
*/
private Integer numWorkers = 1;
+ /**
+ * The maximum capacity of the internal queues. This provides backpressure
+ * to prevent OutOfMemoryErrors when processing is slower than file
listing.
+ */
+ private Integer maxQueueSize = 1000;
+
/**
* If set, do not delete but only rename file that has been processed.
* This config only work when 'keepFile' property is false.
@@ -170,6 +176,12 @@ public class FileSourceConfig implements Serializable {
throw new IllegalArgumentException("The property numWorkers must
be greater than zero");
}
+ // FileSource.open() unboxes maxQueueSize to size the bounded queues,
so an explicit null
+ // would be an NPE there rather than a config error here.
+ if (maxQueueSize == null || maxQueueSize <= 0) {
+ throw new IllegalArgumentException("The property maxQueueSize must
be greater than zero");
+ }
+
if (processedFileSuffix != null && keepFile) {
throw new IllegalArgumentException(
"The property keepFile must be false if the property
processedFileSuffix is set");
diff --git
a/file/src/test/java/org/apache/pulsar/io/file/FileConsumerThreadBoundedQueueTest.java
b/file/src/test/java/org/apache/pulsar/io/file/FileConsumerThreadBoundedQueueTest.java
new file mode 100644
index 00000000..927e8b8b
--- /dev/null
+++
b/file/src/test/java/org/apache/pulsar/io/file/FileConsumerThreadBoundedQueueTest.java
@@ -0,0 +1,90 @@
+/*
+ * 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.mockito.Mockito.any;
+import static org.mockito.Mockito.atLeast;
+import static org.mockito.Mockito.verify;
+import static org.testng.Assert.fail;
+import java.io.File;
+import java.io.IOException;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import org.apache.pulsar.functions.api.Record;
+import org.apache.pulsar.io.core.PushSource;
+import org.awaitility.Awaitility;
+import org.mockito.Mockito;
+import org.testng.annotations.Test;
+
+/**
+ * The consumer must survive a full {@code recentlyProcessed} queue.
+ *
+ * <p>{@code BlockingQueue.add()} throws {@link IllegalStateException} when a
bounded queue is
+ * full — it does not return {@code false}, so the surrounding {@code
do/while(!added)} retry loop
+ * never runs. Once the queues are bounded, a cleanup thread slower than the
workers would kill
+ * the consumer thread outright and stall processing. The producers must block
instead.
+ */
+@SuppressWarnings("unchecked")
+public class FileConsumerThreadBoundedQueueTest extends AbstractFileTest {
+
+ /**
+ * With no cleanup thread draining {@code recentlyProcessed}, the consumer
fills it and must
+ * then block rather than throw. Draining it lets the remaining files
through.
+ */
+ @Test
+ public final void
consumerBlocksInsteadOfCrashingWhenDownstreamQueueIsFull() throws IOException {
+ PushSource<byte[]> consumer = Mockito.mock(PushSource.class);
+ Mockito.doNothing().when(consumer).consume((Record<byte[]>)
any(Record.class));
+
+ BlockingQueue<File> work = new LinkedBlockingQueue<>();
+ BlockingQueue<File> inProcess = new LinkedBlockingQueue<>(10);
+ // Nothing drains this, so it fills after one file.
+ BlockingQueue<File> recentlyProcessed = new LinkedBlockingQueue<>(1);
+
+ try {
+ generateFiles(3);
+ for (File f : producedFiles) {
+ work.put(f);
+ }
+
+ executor.execute(new FileConsumerThread(consumer, work, inProcess,
recentlyProcessed));
+
+ // The first file lands in recentlyProcessed; the consumer then
blocks on the second.
+ Awaitility.await().atMost(10, TimeUnit.SECONDS)
+ .until(() -> recentlyProcessed.size() == 1);
+
+ // Drain as the cleanup thread would. Every file must be handed
over: the consumer
+ // must still be alive, having blocked rather than thrown
IllegalStateException.
+ for (int i = 0; i < 3; i++) {
+ File taken = recentlyProcessed.poll(10, TimeUnit.SECONDS);
+ if (taken == null) {
+ fail("consumer thread died: expected 3 files through
recentlyProcessed, got " + i);
+ }
+ }
+
+ verify(consumer, atLeast(3)).consume(any(Record.class));
+ } catch (InterruptedException | ExecutionException e) {
+ fail("Unable to generate files" + e.getLocalizedMessage());
+ } finally {
+ cleanUp();
+ }
+ }
+}
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 6f2a182e..238fc284 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
@@ -58,16 +58,16 @@ public class FileConsumerThreadTest extends
AbstractFileTest {
for (File produced : producedFiles) {
verify(workQueue, times(1)).offer(produced);
- verify(inProcess, times(1)).add(produced);
+ verify(inProcess, times(1)).put(produced);
verify(inProcess, times(1)).remove(produced);
- verify(recentlyProcessed, times(1)).add(produced);
+ verify(recentlyProcessed, times(1)).put(produced);
}
verify(workQueue, times(1)).offer(any(File.class));
verify(workQueue, atLeast(1)).take();
- verify(inProcess, times(1)).add(any(File.class));
+ verify(inProcess, times(1)).put(any(File.class));
verify(inProcess, times(1)).remove(any(File.class));
- verify(recentlyProcessed, times(1)).add(any(File.class));
+ verify(recentlyProcessed, times(1)).put(any(File.class));
verify(consumer, times(1)).consume((Record<byte[]>)
any(Record.class));
} catch (InterruptedException | ExecutionException e) {
fail("Unable to generate files" + e.getLocalizedMessage());
@@ -93,16 +93,16 @@ public class FileConsumerThreadTest extends
AbstractFileTest {
for (File produced : producedFiles) {
verify(workQueue, times(1)).offer(produced);
- verify(inProcess, times(1)).add(produced);
+ verify(inProcess, times(1)).put(produced);
verify(inProcess, times(1)).remove(produced);
- verify(recentlyProcessed, times(1)).add(produced);
+ verify(recentlyProcessed, times(1)).put(produced);
}
verify(workQueue, times(50)).offer(any(File.class));
verify(workQueue, atLeast(50)).take();
- verify(inProcess, times(50)).add(any(File.class));
+ verify(inProcess, times(50)).put(any(File.class));
verify(inProcess, times(50)).remove(any(File.class));
- verify(recentlyProcessed, times(50)).add(any(File.class));
+ verify(recentlyProcessed, times(50)).put(any(File.class));
verify(consumer, times(100)).consume((Record<byte[]>)
any(Record.class));
} catch (InterruptedException | ExecutionException e) {
fail("Unable to generate files" + e.getLocalizedMessage());
@@ -128,16 +128,16 @@ public class FileConsumerThreadTest extends
AbstractFileTest {
for (File produced : producedFiles) {
verify(workQueue, times(1)).offer(produced);
- verify(inProcess, times(1)).add(produced);
+ verify(inProcess, times(1)).put(produced);
verify(inProcess, times(1)).remove(produced);
- verify(recentlyProcessed, times(1)).add(produced);
+ verify(recentlyProcessed, times(1)).put(produced);
}
verify(workQueue, times(1)).offer(any(File.class));
verify(workQueue, atLeast(1)).take();
- verify(inProcess, times(1)).add(any(File.class));
+ verify(inProcess, times(1)).put(any(File.class));
verify(inProcess, times(1)).remove(any(File.class));
- verify(recentlyProcessed, times(1)).add(any(File.class));
+ verify(recentlyProcessed, times(1)).put(any(File.class));
verify(consumer, times(10)).consume((Record<byte[]>)
any(Record.class));
} catch (InterruptedException | ExecutionException e) {
fail("Unable to generate files" + e.getLocalizedMessage());
diff --git
a/file/src/test/java/org/apache/pulsar/io/file/FileSourceConfigTest.java
b/file/src/test/java/org/apache/pulsar/io/file/FileSourceConfigTest.java
index e7d497f5..20df1451 100644
--- a/file/src/test/java/org/apache/pulsar/io/file/FileSourceConfigTest.java
+++ b/file/src/test/java/org/apache/pulsar/io/file/FileSourceConfigTest.java
@@ -18,8 +18,10 @@
*/
package org.apache.pulsar.io.file;
+import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertThrows;
import static org.testng.Assert.assertTrue;
import java.io.File;
import java.io.IOException;
@@ -142,6 +144,69 @@ public class FileSourceConfigTest {
config.validate();
}
+ @Test
+ public void testDefaultMaxQueueSize() throws Exception {
+ Map<String, Object> map = new HashMap<>();
+ map.put("inputDirectory", INPUT_DIRECTORY);
+
+ FileSourceConfig config = FileSourceConfig.load(map);
+
+ // Assert that if the user provides no value, it defaults to 1000
+ assertEquals(config.getMaxQueueSize(), Integer.valueOf(1000));
+ }
+
+ @Test
+ public void testValidMaxQueueSize() throws Exception {
+ Map<String, Object> map = new HashMap<>();
+ map.put("inputDirectory", INPUT_DIRECTORY);
+ map.put("maxQueueSize", 5000);
+
+ FileSourceConfig config = FileSourceConfig.load(map);
+
+ // Assert that the custom value is loaded correctly
+ assertEquals(config.getMaxQueueSize(), Integer.valueOf(5000));
+
+ // Should not throw any exceptions
+ config.validate();
+ }
+
+ @Test
+ public void testInvalidMaxQueueSize() {
+ Map<String, Object> map = new HashMap<>();
+ map.put("inputDirectory", INPUT_DIRECTORY);
+
+ // Test 0
+ map.put("maxQueueSize", 0);
+ assertThrows(IllegalArgumentException.class, () -> {
+ FileSourceConfig config = FileSourceConfig.load(map);
+ config.validate();
+ });
+
+ // Test Negative Number
+ map.put("maxQueueSize", -5);
+ assertThrows(IllegalArgumentException.class, () -> {
+ FileSourceConfig config = FileSourceConfig.load(map);
+ config.validate();
+ });
+ }
+
+ /**
+ * An explicit null must be rejected by validate(), not deferred to
FileSource.open(),
+ * which unboxes maxQueueSize to size the bounded queues and would throw a
+ * NullPointerException instead of a useful config error.
+ */
+ @Test
+ public void testNullMaxQueueSizeRejected() {
+ Map<String, Object> map = new HashMap<>();
+ map.put("inputDirectory", INPUT_DIRECTORY);
+ map.put("maxQueueSize", null);
+
+ assertThrows(IllegalArgumentException.class, () -> {
+ FileSourceConfig config = FileSourceConfig.load(map);
+ config.validate();
+ });
+ }
+
private File getFile(String name) {
ClassLoader classLoader = getClass().getClassLoader();
return new File(classLoader.getResource(name).getFile());
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 4d109cb7..2594948b 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
@@ -90,16 +90,16 @@ public class ProcessedFileThreadTest extends
AbstractFileTest {
for (File produced : producedFiles) {
verify(workQueue, times(1)).offer(produced);
- verify(inProcess, times(1)).add(produced);
+ verify(inProcess, times(1)).put(produced);
verify(inProcess, times(1)).remove(produced);
- verify(recentlyProcessed, times(1)).add(produced);
+ verify(recentlyProcessed, times(1)).put(produced);
}
verify(workQueue, times(1)).offer(any(File.class));
verify(workQueue, atLeast(1)).take();
- verify(inProcess, times(1)).add(any(File.class));
+ verify(inProcess, times(1)).put(any(File.class));
verify(inProcess, times(1)).remove(any(File.class));
- verify(recentlyProcessed, times(1)).add(any(File.class));
+ verify(recentlyProcessed, times(1)).put(any(File.class));
verify(recentlyProcessed, times(2)).take();
} catch (InterruptedException | ExecutionException e) {
fail("Unable to generate files" + e.getLocalizedMessage());
@@ -129,17 +129,17 @@ public class ProcessedFileThreadTest extends
AbstractFileTest {
for (File produced : producedFiles) {
verify(workQueue, times(1)).offer(produced);
- verify(inProcess, times(1)).add(produced);
+ verify(inProcess, times(1)).put(produced);
verify(inProcess, times(1)).remove(produced);
- verify(recentlyProcessed, times(1)).add(produced);
+ verify(recentlyProcessed, times(1)).put(produced);
}
verify(workQueue, times(50)).offer(any(File.class));
verify(workQueue, atLeast(50)).take();
- verify(inProcess, times(50)).add(any(File.class));
+ verify(inProcess, times(50)).put(any(File.class));
verify(inProcess, times(50)).remove(any(File.class));
- verify(recentlyProcessed, times(50)).add(any(File.class));
- verify(recentlyProcessed, times(50)).add(any(File.class));
+ verify(recentlyProcessed, times(50)).put(any(File.class));
+ verify(recentlyProcessed, times(50)).put(any(File.class));
verify(recentlyProcessed, times(51)).take();
} catch (InterruptedException | ExecutionException e) {
fail("Unable to generate files" + e.getLocalizedMessage());
@@ -170,9 +170,9 @@ public class ProcessedFileThreadTest extends
AbstractFileTest {
for (File produced : producedFiles) {
verify(workQueue, atLeast(4)).offer(produced);
- verify(inProcess, atLeast(4)).add(produced);
+ verify(inProcess, atLeast(4)).put(produced);
verify(inProcess, atLeast(4)).remove(produced);
- verify(recentlyProcessed, atLeast(4)).add(produced);
+ verify(recentlyProcessed, atLeast(4)).put(produced);
}
verify(recentlyProcessed, atLeast(5)).take();
@@ -218,9 +218,9 @@ public class ProcessedFileThreadTest extends
AbstractFileTest {
// Make sure every single file was processed.
for (File produced : producedFiles) {
verify(workQueue, times(1)).offer(produced);
- verify(inProcess, times(1)).add(produced);
+ verify(inProcess, times(1)).put(produced);
verify(inProcess, times(1)).remove(produced);
- verify(recentlyProcessed, times(1)).add(produced);
+ verify(recentlyProcessed, times(1)).put(produced);
}
} catch (InterruptedException e) {
@@ -271,9 +271,9 @@ public class ProcessedFileThreadTest extends
AbstractFileTest {
// Make sure every single file was processed exactly once.
for (File produced : producedFiles) {
verify(workQueue, times(1)).offer(produced);
- verify(inProcess, times(1)).add(produced);
+ verify(inProcess, times(1)).put(produced);
verify(inProcess, times(1)).remove(produced);
- verify(recentlyProcessed, times(1)).add(produced);
+ verify(recentlyProcessed, times(1)).put(produced);
}
} catch (InterruptedException e) {
@@ -322,9 +322,9 @@ public class ProcessedFileThreadTest extends
AbstractFileTest {
// Make sure every single file was processed.
for (File produced : producedFiles) {
verify(workQueue, times(1)).offer(produced);
- verify(inProcess, times(1)).add(produced);
+ verify(inProcess, times(1)).put(produced);
verify(inProcess, times(1)).remove(produced);
- verify(recentlyProcessed, times(1)).add(produced);
+ verify(recentlyProcessed, times(1)).put(produced);
assert(!produced.exists());
assert(new File(produced.getAbsolutePath() +
processedFileSuffix).exists());
