This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new a5ee1ba7fcbc camel-file-watch: report events dropped by a full event
queue (#29680)
a5ee1ba7fcbc is described below
commit a5ee1ba7fcbc422298822d6432f478f7f5c8d0d8
Author: dhairya05 <[email protected]>
AuthorDate: Thu Nov 12 17:32:06 2026 +0530
camel-file-watch: report events dropped by a full event queue (#29680)
* camel-file-watch: report events dropped by a full event queue
The watcher thread handed every matched event to the queue the polling
threads read from, and discarded the result:
eventQueue.offer(fileEvent);
eventQueue is bounded whenever the queueSize option is set, so offer()
returns false once it is full and the event is lost without a trace - no
exchange, no log line, nothing to tell the user that queueSize is too small
for their rate of changes. The branch directly above warns about exactly
this
kind of loss for the watch service's own OVERFLOW event.
The call moves into offerEvent(), which logs a warning naming the event, the
path and the configured size when the queue rejects it.
* camel-file-watch: warn once about a full queue, and document the drop
Review follow-up on #29680.
One WARN per dropped event would flood the log exactly when the consumer is
already behind: a full queue means a burst, so a single burst could produce
thousands of identical lines and the extra log I/O makes the backlog worse.
The first drop now warns with the configured size, the offending event and
the hint, and says that further drops are logged at DEBUG; the rest go to
DEBUG. The flag is reset in doStart so a restarted consumer warns again.
Every drop is still reported to the caller, which the test now covers with a
third event.
The queueSize description said only "Maximum size of queue between
WatchService and consumer. Unbounded by default." - it did not say what
happens when the queue fills. It now does, propagated to the component and
catalog JSON and to both DSL builders.
Dropped the unbounded-queue test: it only exercised LinkedBlockingQueue's
default capacity.
---
.../camel/catalog/components/file-watch.json | 4 +-
.../camel/component/file/watch/file-watch.json | 4 +-
.../component/file/watch/FileWatchComponent.java | 3 +-
.../component/file/watch/FileWatchConsumer.java | 31 ++++++++++-
.../component/file/watch/FileWatchEndpoint.java | 3 +-
.../file/watch/FileWatchConsumerQueueTest.java | 60 ++++++++++++++++++++++
.../dsl/FileWatchComponentBuilderFactory.java | 2 +-
.../dsl/FileWatchEndpointBuilderFactory.java | 4 +-
8 files changed, 101 insertions(+), 10 deletions(-)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/file-watch.json
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/file-watch.json
index 641aec2ce7f4..6f1dcd589a51 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/file-watch.json
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/components/file-watch.json
@@ -30,7 +30,7 @@
"concurrentConsumers": { "index": 3, "kind": "property", "displayName":
"Concurrent Consumers", "group": "advanced", "label": "advanced", "required":
false, "type": "integer", "javaType": "int", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": 1, "description": "The number of
concurrent consumers. Increase this value, if your route is slow to prevent
buffering in queue." },
"fileHasher": { "index": 4, "kind": "property", "displayName": "File
Hasher", "group": "advanced", "label": "advanced", "required": false, "type":
"object", "javaType": "io.methvin.watcher.hashing.FileHasher", "deprecated":
false, "autowired": false, "secret": false, "defaultValue":
"#murmur3FFileHasher", "description": "Reference to
io.methvin.watcher.hashing.FileHasher. This prevents emitting duplicate events
on some platforms. For working with large files and if you dont need dete [...]
"pollThreads": { "index": 5, "kind": "property", "displayName": "Poll
Threads", "group": "advanced", "label": "advanced", "required": false, "type":
"integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 1, "description": "The number of threads
polling WatchService. Increase this value, if you see OVERFLOW messages in
log." },
- "queueSize": { "index": 6, "kind": "property", "displayName": "Queue
Size", "group": "advanced", "label": "advanced", "required": false, "type":
"integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 2147483647, "description": "Maximum size of
queue between WatchService and consumer. Unbounded by default." }
+ "queueSize": { "index": 6, "kind": "property", "displayName": "Queue
Size", "group": "advanced", "label": "advanced", "required": false, "type":
"integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 2147483647, "description": "Maximum size of
queue between WatchService and consumer. Unbounded by default. When the queue
is full, file events are dropped." }
},
"headers": {
"CamelFileEventType": { "index": 0, "kind": "header", "displayName": "",
"group": "consumer", "label": "", "required": false, "javaType": "String",
"deprecated": false, "autowired": false, "secret": false, "description": "Type
of event. Possible values: CREATE, DELETE, MODIFY.", "constantName":
"org.apache.camel.component.file.watch.FileWatchConstants#EVENT_TYPE_HEADER" },
@@ -60,6 +60,6 @@
"concurrentConsumers": { "index": 9, "kind": "parameter", "displayName":
"Concurrent Consumers", "group": "advanced", "label": "advanced", "required":
false, "type": "integer", "javaType": "int", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": 1, "description": "The number of
concurrent consumers. Increase this value, if your route is slow to prevent
buffering in queue." },
"fileHasher": { "index": 10, "kind": "parameter", "displayName": "File
Hasher", "group": "advanced", "label": "advanced", "required": false, "type":
"object", "javaType": "io.methvin.watcher.hashing.FileHasher", "deprecated":
false, "autowired": false, "secret": false, "defaultValue":
"#murmur3FFileHasher", "description": "Reference to
io.methvin.watcher.hashing.FileHasher. This prevents emitting duplicate events
on some platforms. For working with large files and if you dont need de [...]
"pollThreads": { "index": 11, "kind": "parameter", "displayName": "Poll
Threads", "group": "advanced", "label": "advanced", "required": false, "type":
"integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 1, "description": "The number of threads
polling WatchService. Increase this value, if you see OVERFLOW messages in
log." },
- "queueSize": { "index": 12, "kind": "parameter", "displayName": "Queue
Size", "group": "advanced", "label": "advanced", "required": false, "type":
"integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 2147483647, "description": "Maximum size of
queue between WatchService and consumer. Unbounded by default." }
+ "queueSize": { "index": 12, "kind": "parameter", "displayName": "Queue
Size", "group": "advanced", "label": "advanced", "required": false, "type":
"integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 2147483647, "description": "Maximum size of
queue between WatchService and consumer. Unbounded by default. When the queue
is full, file events are dropped." }
}
}
diff --git
a/components/camel-file-watch/src/generated/resources/META-INF/org/apache/camel/component/file/watch/file-watch.json
b/components/camel-file-watch/src/generated/resources/META-INF/org/apache/camel/component/file/watch/file-watch.json
index 641aec2ce7f4..6f1dcd589a51 100644
---
a/components/camel-file-watch/src/generated/resources/META-INF/org/apache/camel/component/file/watch/file-watch.json
+++
b/components/camel-file-watch/src/generated/resources/META-INF/org/apache/camel/component/file/watch/file-watch.json
@@ -30,7 +30,7 @@
"concurrentConsumers": { "index": 3, "kind": "property", "displayName":
"Concurrent Consumers", "group": "advanced", "label": "advanced", "required":
false, "type": "integer", "javaType": "int", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": 1, "description": "The number of
concurrent consumers. Increase this value, if your route is slow to prevent
buffering in queue." },
"fileHasher": { "index": 4, "kind": "property", "displayName": "File
Hasher", "group": "advanced", "label": "advanced", "required": false, "type":
"object", "javaType": "io.methvin.watcher.hashing.FileHasher", "deprecated":
false, "autowired": false, "secret": false, "defaultValue":
"#murmur3FFileHasher", "description": "Reference to
io.methvin.watcher.hashing.FileHasher. This prevents emitting duplicate events
on some platforms. For working with large files and if you dont need dete [...]
"pollThreads": { "index": 5, "kind": "property", "displayName": "Poll
Threads", "group": "advanced", "label": "advanced", "required": false, "type":
"integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 1, "description": "The number of threads
polling WatchService. Increase this value, if you see OVERFLOW messages in
log." },
- "queueSize": { "index": 6, "kind": "property", "displayName": "Queue
Size", "group": "advanced", "label": "advanced", "required": false, "type":
"integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 2147483647, "description": "Maximum size of
queue between WatchService and consumer. Unbounded by default." }
+ "queueSize": { "index": 6, "kind": "property", "displayName": "Queue
Size", "group": "advanced", "label": "advanced", "required": false, "type":
"integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 2147483647, "description": "Maximum size of
queue between WatchService and consumer. Unbounded by default. When the queue
is full, file events are dropped." }
},
"headers": {
"CamelFileEventType": { "index": 0, "kind": "header", "displayName": "",
"group": "consumer", "label": "", "required": false, "javaType": "String",
"deprecated": false, "autowired": false, "secret": false, "description": "Type
of event. Possible values: CREATE, DELETE, MODIFY.", "constantName":
"org.apache.camel.component.file.watch.FileWatchConstants#EVENT_TYPE_HEADER" },
@@ -60,6 +60,6 @@
"concurrentConsumers": { "index": 9, "kind": "parameter", "displayName":
"Concurrent Consumers", "group": "advanced", "label": "advanced", "required":
false, "type": "integer", "javaType": "int", "deprecated": false, "autowired":
false, "secret": false, "defaultValue": 1, "description": "The number of
concurrent consumers. Increase this value, if your route is slow to prevent
buffering in queue." },
"fileHasher": { "index": 10, "kind": "parameter", "displayName": "File
Hasher", "group": "advanced", "label": "advanced", "required": false, "type":
"object", "javaType": "io.methvin.watcher.hashing.FileHasher", "deprecated":
false, "autowired": false, "secret": false, "defaultValue":
"#murmur3FFileHasher", "description": "Reference to
io.methvin.watcher.hashing.FileHasher. This prevents emitting duplicate events
on some platforms. For working with large files and if you dont need de [...]
"pollThreads": { "index": 11, "kind": "parameter", "displayName": "Poll
Threads", "group": "advanced", "label": "advanced", "required": false, "type":
"integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 1, "description": "The number of threads
polling WatchService. Increase this value, if you see OVERFLOW messages in
log." },
- "queueSize": { "index": 12, "kind": "parameter", "displayName": "Queue
Size", "group": "advanced", "label": "advanced", "required": false, "type":
"integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 2147483647, "description": "Maximum size of
queue between WatchService and consumer. Unbounded by default." }
+ "queueSize": { "index": 12, "kind": "parameter", "displayName": "Queue
Size", "group": "advanced", "label": "advanced", "required": false, "type":
"integer", "javaType": "int", "deprecated": false, "autowired": false,
"secret": false, "defaultValue": 2147483647, "description": "Maximum size of
queue between WatchService and consumer. Unbounded by default. When the queue
is full, file events are dropped." }
}
}
diff --git
a/components/camel-file-watch/src/main/java/org/apache/camel/component/file/watch/FileWatchComponent.java
b/components/camel-file-watch/src/main/java/org/apache/camel/component/file/watch/FileWatchComponent.java
index 7f05fe4cc78c..8f16dc9ebb05 100644
---
a/components/camel-file-watch/src/main/java/org/apache/camel/component/file/watch/FileWatchComponent.java
+++
b/components/camel-file-watch/src/main/java/org/apache/camel/component/file/watch/FileWatchComponent.java
@@ -37,7 +37,8 @@ public class FileWatchComponent extends DefaultComponent {
private int concurrentConsumers = 1;
@Metadata(label = "advanced",
- description = "Maximum size of queue between WatchService and
consumer. Unbounded by default.",
+ description = "Maximum size of queue between WatchService and
consumer. Unbounded by default."
+ + " When the queue is full, file events are
dropped.",
defaultValue = "" + Integer.MAX_VALUE)
private int queueSize = Integer.MAX_VALUE;
diff --git
a/components/camel-file-watch/src/main/java/org/apache/camel/component/file/watch/FileWatchConsumer.java
b/components/camel-file-watch/src/main/java/org/apache/camel/component/file/watch/FileWatchConsumer.java
index ecff70145fab..089c67066a64 100644
---
a/components/camel-file-watch/src/main/java/org/apache/camel/component/file/watch/FileWatchConsumer.java
+++
b/components/camel-file-watch/src/main/java/org/apache/camel/component/file/watch/FileWatchConsumer.java
@@ -26,6 +26,7 @@ import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
import io.methvin.watcher.DirectoryChangeEvent;
import io.methvin.watcher.DirectoryChangeListener;
@@ -55,6 +56,7 @@ public class FileWatchConsumer extends DefaultConsumer {
private Path baseDirectory;
private AntPathMatcher antPathMatcher;
private DirectoryWatcher watcher;
+ private final AtomicBoolean queueFullWarned = new AtomicBoolean();
public FileWatchConsumer(FileWatchEndpoint endpoint, Processor processor) {
super(endpoint, processor);
@@ -82,6 +84,9 @@ public class FileWatchConsumer extends DefaultConsumer {
protected void doStart() throws Exception {
super.doStart();
+ // so that a restarted consumer warns again about a queue that fills up
+ queueFullWarned.set(false);
+
if (!Files.exists(baseDirectory)) {
if (getEndpoint().isAutoCreate()) {
baseDirectory = Files.createDirectories(baseDirectory);
@@ -224,6 +229,30 @@ public class FileWatchConsumer extends DefaultConsumer {
return (FileWatchEndpoint) super.getEndpoint();
}
+ /**
+ * Hands a matched event over to the polling threads.
+ *
+ * @param fileEvent the event to queue
+ * @return {@code false} when the queue was full and the event
was dropped
+ */
+ boolean offerEvent(FileEvent fileEvent) {
+ if (eventQueue.offer(fileEvent)) {
+ return true;
+ }
+
+ // a full queue means a burst the polling threads are behind on, so
warn once rather than once per lost
+ // event: thousands of identical warnings would add log I/O exactly
when the consumer is already overloaded
+ if (queueFullWarned.compareAndSet(false, true)) {
+ LOG.warn("The event queue is full at its configured size of {}, so
file events are dropped (first: {} {})."
+ + " Consider increasing the option 'queueSize'. Further
dropped events are logged at DEBUG level.",
+ getEndpoint().getQueueSize(), fileEvent.getEventType(),
fileEvent.getEventPath());
+ } else {
+ LOG.debug("Dropping {} event for {}: the event queue is full",
+ fileEvent.getEventType(), fileEvent.getEventPath());
+ }
+ return false;
+ }
+
class FileWatchDirectoryChangeListener implements DirectoryChangeListener {
@Override
public void onEvent(DirectoryChangeEvent directoryChangeEvent) {
@@ -233,7 +262,7 @@ public class FileWatchConsumer extends DefaultConsumer {
}
FileEvent fileEvent = new FileEvent(directoryChangeEvent);
if (matchFilters(fileEvent)) {
- eventQueue.offer(fileEvent);
+ offerEvent(fileEvent);
}
}
diff --git
a/components/camel-file-watch/src/main/java/org/apache/camel/component/file/watch/FileWatchEndpoint.java
b/components/camel-file-watch/src/main/java/org/apache/camel/component/file/watch/FileWatchEndpoint.java
index 742dd9ba72ff..d2d184edad5d 100644
---
a/components/camel-file-watch/src/main/java/org/apache/camel/component/file/watch/FileWatchEndpoint.java
+++
b/components/camel-file-watch/src/main/java/org/apache/camel/component/file/watch/FileWatchEndpoint.java
@@ -69,7 +69,8 @@ public class FileWatchEndpoint extends DefaultEndpoint
implements MultipleConsum
defaultValue = "**")
private String antInclude;
@UriParam(label = "advanced",
- description = "Maximum size of queue between WatchService and
consumer. Unbounded by default.",
+ description = "Maximum size of queue between WatchService and
consumer. Unbounded by default."
+ + " When the queue is full, file events are
dropped.",
defaultValue = "" + Integer.MAX_VALUE)
private int queueSize;
@UriParam(label = "advanced",
diff --git
a/components/camel-file-watch/src/test/java/org/apache/camel/component/file/watch/FileWatchConsumerQueueTest.java
b/components/camel-file-watch/src/test/java/org/apache/camel/component/file/watch/FileWatchConsumerQueueTest.java
new file mode 100644
index 000000000000..d570e3cf0c8b
--- /dev/null
+++
b/components/camel-file-watch/src/test/java/org/apache/camel/component/file/watch/FileWatchConsumerQueueTest.java
@@ -0,0 +1,60 @@
+/*
+ * 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.camel.component.file.watch;
+
+import java.nio.file.Path;
+
+import org.apache.camel.component.file.watch.constants.FileEvent;
+import org.apache.camel.component.file.watch.constants.FileEventEnum;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The queue between the watch service and the consumer is bounded when the
{@code queueSize} option is set, so an event
+ * can fail to be queued. That has to be reported rather than discarded
silently.
+ */
+public class FileWatchConsumerQueueTest {
+
+ @TempDir
+ Path folder;
+
+ @Test
+ public void shouldReportAnEventDroppedByAFullQueue() throws Exception {
+ try (DefaultCamelContext context = new DefaultCamelContext()) {
+ context.start();
+
+ FileWatchEndpoint endpoint = context.getEndpoint(
+ "file-watch:" + folder.toAbsolutePath() + "?queueSize=1",
FileWatchEndpoint.class);
+ FileWatchConsumer consumer = (FileWatchConsumer)
endpoint.createConsumer(exchange -> {
+ });
+
+ FileEvent first = new FileEvent(FileEventEnum.CREATE,
folder.resolve("first.txt"));
+ FileEvent second = new FileEvent(FileEventEnum.CREATE,
folder.resolve("second.txt"));
+ FileEvent third = new FileEvent(FileEventEnum.CREATE,
folder.resolve("third.txt"));
+
+ assertTrue(consumer.offerEvent(first), "the first event fits in a
queue of size 1");
+ assertFalse(consumer.offerEvent(second), "the second event is
dropped, and that has to be reported");
+ // the warning is only logged for the first drop, but every drop
still has to be reported to the caller
+ assertFalse(consumer.offerEvent(third), "a later drop is reported
too, even though it only logs at DEBUG");
+ }
+ }
+
+}
diff --git
a/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/FileWatchComponentBuilderFactory.java
b/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/FileWatchComponentBuilderFactory.java
index bd5742400438..eca5a0cae7b1 100644
---
a/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/FileWatchComponentBuilderFactory.java
+++
b/dsl/camel-componentdsl/src/generated/java/org/apache/camel/builder/component/dsl/FileWatchComponentBuilderFactory.java
@@ -183,7 +183,7 @@ public interface FileWatchComponentBuilderFactory {
/**
* Maximum size of queue between WatchService and consumer. Unbounded
by
- * default.
+ * default. When the queue is full, file events are dropped.
*
* The option is a: <code>int</code> type.
*
diff --git
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/FileWatchEndpointBuilderFactory.java
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/FileWatchEndpointBuilderFactory.java
index 5dc597406743..319e8fe71e79 100644
---
a/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/FileWatchEndpointBuilderFactory.java
+++
b/dsl/camel-endpointdsl/src/generated/java/org/apache/camel/builder/endpoint/dsl/FileWatchEndpointBuilderFactory.java
@@ -411,7 +411,7 @@ public interface FileWatchEndpointBuilderFactory {
}
/**
* Maximum size of queue between WatchService and consumer. Unbounded
by
- * default.
+ * default. When the queue is full, file events are dropped.
*
* The option is a: <code>int</code> type.
*
@@ -427,7 +427,7 @@ public interface FileWatchEndpointBuilderFactory {
}
/**
* Maximum size of queue between WatchService and consumer. Unbounded
by
- * default.
+ * default. When the queue is full, file events are dropped.
*
* The option will be converted to a <code>int</code> type.
*
