This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch camel-4.18.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.18.x by this push:
new aaeaead8429f [backport camel-4.18.x] CAMEL-25930: camel-file -
GenericFilePollingConsumer should return the polled exchange directly from
receive (#28267)
aaeaead8429f is described below
commit aaeaead8429fb36e48682d2c03a4a7d9f66abc80
Author: Guillaume Nodet <[email protected]>
AuthorDate: Thu Oct 22 09:16:20 2026 +0200
[backport camel-4.18.x] CAMEL-25930: camel-file -
GenericFilePollingConsumer should return the polled exchange directly from
receive (#28267)
GenericFilePollingConsumer handed the polled exchange from process() to
pollAndTake() through a shared BlockingQueue, so with concurrent callers of
receive() a thread could take the exchange polled by another thread. The
polled exchange is now returned directly from receive.
Backport of #28248.
Co-Authored-By: Claude Opus 5.5 (1M context) <[email protected]>
---
.../component/file/GenericFilePollingConsumer.java | 52 ++++++++-----
.../file/FilePollEnrichDynamicFileNameTest.java | 87 ++++++++++++++++++++++
2 files changed, 121 insertions(+), 18 deletions(-)
diff --git
a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFilePollingConsumer.java
b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFilePollingConsumer.java
index 9289d431ede6..4cd03b0a37ca 100644
---
a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFilePollingConsumer.java
+++
b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFilePollingConsumer.java
@@ -19,6 +19,7 @@ package org.apache.camel.component.file;
import org.apache.camel.Consumer;
import org.apache.camel.DynamicPollingConsumer;
import org.apache.camel.Exchange;
+import org.apache.camel.PooledExchange;
import org.apache.camel.RuntimeCamelException;
import org.apache.camel.spi.PollingConsumerPollStrategy;
import org.apache.camel.support.EventDrivenPollingConsumer;
@@ -31,6 +32,8 @@ public class GenericFilePollingConsumer extends
EventDrivenPollingConsumer imple
private static final Logger LOG =
LoggerFactory.getLogger(GenericFilePollingConsumer.class);
private final long delay;
+ // the exchange found by the poll of the calling thread, returned directly
by receive
+ private final ThreadLocal<Exchange[]> polledByThisThread = new
ThreadLocal<>();
public GenericFilePollingConsumer(GenericFileEndpoint endpoint) {
super(endpoint);
@@ -73,12 +76,7 @@ public class GenericFilePollingConsumer extends
EventDrivenPollingConsumer imple
if (LOG.isTraceEnabled()) {
LOG.trace("receiveNoWait polling file: {}",
getConsumer().getEndpoint());
}
- int polled = doReceive(exchange, 0);
- if (polled > 0) {
- return super.receive(0);
- } else {
- return null;
- }
+ return pollAndTake(exchange, 0, 0);
}
@Override
@@ -86,12 +84,7 @@ public class GenericFilePollingConsumer extends
EventDrivenPollingConsumer imple
if (LOG.isTraceEnabled()) {
LOG.trace("receive polling file: {}", getConsumer().getEndpoint());
}
- int polled = doReceive(exchange, Long.MAX_VALUE);
- if (polled > 0) {
- return super.receive();
- } else {
- return null;
- }
+ return pollAndTake(exchange, Long.MAX_VALUE, -1);
}
@Override
@@ -99,12 +92,7 @@ public class GenericFilePollingConsumer extends
EventDrivenPollingConsumer imple
if (LOG.isTraceEnabled()) {
LOG.trace("receive({}) polling file: {}", timeout,
getConsumer().getEndpoint());
}
- int polled = doReceive(exchange, timeout);
- if (polled > 0) {
- return super.receive(timeout);
- } else {
- return null;
- }
+ return pollAndTake(exchange, timeout, timeout);
}
@Override
@@ -122,6 +110,28 @@ public class GenericFilePollingConsumer extends
EventDrivenPollingConsumer imple
return receive(null, timeout);
}
+ /**
+ * Polls and returns the exchange found by this poll, or the head of the
queue when the exchange was processed on
+ * another thread.
+ */
+ private Exchange pollAndTake(Exchange exchange, long pollTimeout, long
receiveTimeout) {
+ Exchange[] slot = new Exchange[1];
+ polledByThisThread.set(slot);
+ int polled;
+ try {
+ polled = doReceive(exchange, pollTimeout);
+ } finally {
+ polledByThisThread.remove();
+ }
+ if (slot[0] != null) {
+ return slot[0];
+ } else if (polled > 0) {
+ return receiveTimeout < 0 ? super.receive() :
super.receive(receiveTimeout);
+ } else {
+ return null;
+ }
+ }
+
protected int doReceive(Exchange exchange, long timeout) {
int retryCounter = -1;
boolean done = false;
@@ -217,6 +227,12 @@ public class GenericFilePollingConsumer extends
EventDrivenPollingConsumer imple
if (name != null) {
LOG.debug("Received file: {}", name);
}
+ Exchange[] slot = polledByThisThread.get();
+ if (slot != null && slot[0] == null) {
+ // a receive call on this thread is polling, so return the
exchange to that call
+ slot[0] = isCopy() || exchange instanceof PooledExchange ?
prepareCopy(exchange, true) : exchange;
+ return;
+ }
super.process(exchange);
}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/component/file/FilePollEnrichDynamicFileNameTest.java
b/core/camel-core/src/test/java/org/apache/camel/component/file/FilePollEnrichDynamicFileNameTest.java
new file mode 100644
index 000000000000..fd9c4317c06c
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/component/file/FilePollEnrichDynamicFileNameTest.java
@@ -0,0 +1,87 @@
+/*
+ * 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;
+
+import java.nio.file.Files;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.builder.RouteBuilder;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Poll enrich with a dynamic fileName returns the content of the file named
by each message.
+ */
+public class FilePollEnrichDynamicFileNameTest extends ContextTestSupport {
+
+ private static final int ROUNDS = 20;
+ private static final int FILES = 40;
+ private static final int THREADS = 20;
+
+ @Test
+ public void testPollEnrichDynamicFileName() throws Exception {
+ ExecutorService executor = Executors.newFixedThreadPool(THREADS);
+ List<String> mismatches = new ArrayList<>();
+ try {
+ for (int round = 0; round < ROUNDS; round++) {
+ List<String> names = new ArrayList<>();
+ for (int i = 0; i < FILES; i++) {
+ String name = "r" + round + "-f" + i + ".txt";
+ Files.writeString(testDirectory().resolve(name), "content
of " + name);
+ names.add(name);
+ }
+
+ List<Callable<String>> calls = new ArrayList<>();
+ for (String name : names) {
+ calls.add(() -> {
+ String body =
template.requestBodyAndHeader("direct:start", null, "docName", name,
String.class);
+ return ("content of " + name).equals(body) ? null :
name + " received: " + body;
+ });
+ }
+ for (Future<String> future : executor.invokeAll(calls)) {
+ String mismatch = future.get();
+ if (mismatch != null) {
+ mismatches.add(mismatch);
+ }
+ }
+ }
+ } finally {
+ executor.shutdownNow();
+ }
+
+ assertEquals(List.of(), mismatches);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+
.pollEnrich().simple(fileUri("?noop=true&fileName=${header.docName}")).timeout(5000)
+ .convertBodyTo(String.class);
+ }
+ };
+ }
+}
