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 232c79615dd2 CAMEL-25856: camel-file - keep the lock of a file name
while it is in use (#28379)
232c79615dd2 is described below
commit 232c79615dd27c3b796cb28f361be63895b1440b
Author: allthingssecurity <[email protected]>
AuthorDate: Sun Oct 25 13:00:14 2026 +0530
CAMEL-25856: camel-file - keep the lock of a file name while it is in use
(#28379)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
components/camel-file/pom.xml | 6 +
.../camel/component/file/GenericFileProducer.java | 47 +++---
.../file/FileProducerSameFileLockTest.java | 178 +++++++++++++++++++++
3 files changed, 212 insertions(+), 19 deletions(-)
diff --git a/components/camel-file/pom.xml b/components/camel-file/pom.xml
index 409f05cfaca0..31f573f0338c 100644
--- a/components/camel-file/pom.xml
+++ b/components/camel-file/pom.xml
@@ -54,6 +54,12 @@
<artifactId>camel-core-engine</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <version>${awaitility-version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git
a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileProducer.java
b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileProducer.java
index 30bac75a664f..2502dc7dbe6e 100644
---
a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileProducer.java
+++
b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileProducer.java
@@ -18,6 +18,7 @@ package org.apache.camel.component.file;
import java.io.File;
import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
@@ -26,8 +27,6 @@ import org.apache.camel.Exchange;
import org.apache.camel.Expression;
import org.apache.camel.support.DefaultAsyncProducer;
import org.apache.camel.support.DefaultExchange;
-import org.apache.camel.support.LRUCacheFactory;
-import org.apache.camel.support.service.ServiceHelper;
import org.apache.camel.util.FileUtil;
import org.apache.camel.util.ObjectHelper;
import org.apache.camel.util.StringHelper;
@@ -42,9 +41,10 @@ public class GenericFileProducer<T> extends
DefaultAsyncProducer {
protected final GenericFileEndpoint<T> endpoint;
protected GenericFileOperations<T> operations;
- // assume writing to 100 different files concurrently at most for the same
- // file producer
- private final Map<String, Lock> locks = LRUCacheFactory.newLRUCache(100);
+ // one lock per file name that is being written, so two exchanges do not
write the same file at the same time; a
+ // lock is kept while an exchange holds or waits for it, and removed by
the last one, so it is never replaced by a
+ // new lock while it is in use
+ private final Map<String, FileLock> locks = new ConcurrentHashMap<>();
protected GenericFileProducer(GenericFileEndpoint<T> endpoint,
GenericFileOperations<T> operations) {
super(endpoint);
@@ -82,17 +82,13 @@ public class GenericFileProducer<T> extends
DefaultAsyncProducer {
// use lock for same file name to avoid concurrent writes to the same
// file
// for example when you concurrently append to the same file
- Lock lock = locks.computeIfAbsent(target, f -> new ReentrantLock());
-
- lock.lock();
+ FileLock lock = acquireLock(target);
try {
processExchange(exchange, target);
} catch (Exception e) {
exchange.setException(e);
} finally {
- // do not remove as the locks cache has an upper bound
- // this ensure the locks is appropriate reused
- lock.unlock();
+ releaseLock(target, lock);
// and remove the write file name header as we only want to use it
// once (by design)
exchange.getIn().removeHeader(Exchange.OVERRULE_FILE_NAME);
@@ -513,16 +509,29 @@ public class GenericFileProducer<T> extends
DefaultAsyncProducer {
return answer;
}
- @Override
- protected void doStart() throws Exception {
- ServiceHelper.startService(locks);
- super.doStart();
+ private FileLock acquireLock(String target) {
+ // count the user before taking the lock, so the lock stays in the map
while this exchange waits for it
+ FileLock lock = locks.compute(target, (k, v) -> {
+ FileLock answer = v != null ? v : new FileLock();
+ answer.users++;
+ return answer;
+ });
+ lock.lock.lock();
+ return lock;
}
- @Override
- protected void doStop() throws Exception {
- super.doStop();
- ServiceHelper.stopService(locks);
+ private void releaseLock(String target, FileLock lock) {
+ lock.lock.unlock();
+ locks.computeIfPresent(target, (k, v) -> {
+ v.users--;
+ return v.users == 0 ? null : v;
+ });
+ }
+
+ private static final class FileLock {
+ private final Lock lock = new ReentrantLock();
+ // guarded by the map entry (compute / computeIfPresent)
+ private int users;
}
}
diff --git
a/components/camel-file/src/test/java/org/apache/camel/component/file/FileProducerSameFileLockTest.java
b/components/camel-file/src/test/java/org/apache/camel/component/file/FileProducerSameFileLockTest.java
new file mode 100644
index 000000000000..c3c79ff4ef3d
--- /dev/null
+++
b/components/camel-file/src/test/java/org/apache/camel/component/file/FileProducerSameFileLockTest.java
@@ -0,0 +1,178 @@
+/*
+ * 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.io.InputStream;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.LockSupport;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Endpoint;
+import org.apache.camel.Exchange;
+import org.apache.camel.Producer;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.support.DefaultExchange;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The file producer writes one file name at a time, so that two exchanges
that append to the same file do not mix their
+ * content. That must still hold when the producer has written many other
files while one write was going on.
+ */
+class FileProducerSameFileLockTest {
+
+ @TempDir
+ Path dir;
+
+ private CamelContext context;
+ private Producer producer;
+ private ExecutorService executor;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ // the context is not started: camel-file is tested without the simple
language here
+ context = new DefaultCamelContext();
+ Endpoint endpoint = context.getEndpoint("file:" + dir.toAbsolutePath()
+ "?fileExist=Append&bufferSize=4");
+ producer = endpoint.createProducer();
+ producer.start();
+ executor = Executors.newFixedThreadPool(2);
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ if (executor != null) {
+ executor.shutdownNow();
+ }
+ if (producer != null) {
+ producer.stop();
+ }
+ context.stop();
+ }
+
+ @Test
+ void testSameFileIsWrittenOneAtATime() throws Exception {
+ assertSecondWriterWaits(10);
+ }
+
+ @Test
+ void testSameFileIsWrittenOneAtATimeAfterManyOtherFiles() throws Exception
{
+ // more file names than the 100 the producer used to keep a lock for
+ assertSecondWriterWaits(150);
+ }
+
+ private void assertSecondWriterWaits(int otherFiles) throws Exception {
+ CountDownLatch firstPartWritten = new CountDownLatch(1);
+ CountDownLatch release = new CountDownLatch(1);
+
+ // a slow writer: it appends AAAA, waits, then appends AAAA again
+ Future<Exchange> slow = executor.submit(() -> send("same.txt", new
SlowStream(firstPartWritten, release)));
+ assertTrue(firstPartWritten.await(10, TimeUnit.SECONDS));
+
+ // meanwhile the producer writes other files
+ for (int i = 0; i < otherFiles; i++) {
+ Exchange other = send("other-" + i + ".txt", "x");
+ assertNull(other.getException());
+ }
+
+ // a second writer of the same file must wait until the slow writer is
done: it either parks on a lock (the
+ // lock of the file name) or, without the lock, completes its write
+ FutureTask<Exchange> second = new FutureTask<>(() -> send("same.txt",
"BBBB"));
+ Thread secondWriter = new Thread(second, "second-writer");
+ secondWriter.start();
+ await().atMost(10, TimeUnit.SECONDS)
+ .until(() -> second.isDone() ||
LockSupport.getBlocker(secondWriter) != null);
+ boolean secondDoneWhileFirstWriting = second.isDone();
+ assertTrue(secondDoneWhileFirstWriting ||
LockSupport.getBlocker(secondWriter) != null,
+ "the second write of same.txt neither waits nor completes");
+
+ release.countDown();
+ assertNull(slow.get(10, TimeUnit.SECONDS).getException());
+ assertNull(second.get(10, TimeUnit.SECONDS).getException());
+ assertEquals("AAAAAAAABBBB", Files.readString(dir.resolve("same.txt")),
+ "the two appends to same.txt are mixed");
+ assertFalse(secondDoneWhileFirstWriting, "the second write of same.txt
ran while the first one was writing");
+ }
+
+ private Exchange send(String fileName, Object body) throws Exception {
+ Exchange exchange = new DefaultExchange(context);
+ exchange.getIn().setHeader(Exchange.FILE_NAME, fileName);
+ exchange.getIn().setBody(body);
+ producer.process(exchange);
+ return exchange;
+ }
+
+ private static final class SlowStream extends InputStream {
+ private final CountDownLatch firstPartWritten;
+ private final CountDownLatch release;
+ private int reads;
+
+ SlowStream(CountDownLatch firstPartWritten, CountDownLatch release) {
+ this.firstPartWritten = firstPartWritten;
+ this.release = release;
+ }
+
+ @Override
+ public int read() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public int read(byte[] b) {
+ return read(b, 0, b.length);
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) {
+ reads++;
+ if (reads == 2) {
+ // the first AAAA has been handed to the file; hold the write
here
+ firstPartWritten.countDown();
+ try {
+ if (!release.await(20, TimeUnit.SECONDS)) {
+ throw new IllegalStateException("not released");
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ throw new IllegalStateException(e);
+ }
+ }
+ if (reads > 2) {
+ return -1;
+ }
+ byte[] data = "AAAA".getBytes(StandardCharsets.US_ASCII);
+ int n = Math.min(len, data.length);
+ System.arraycopy(data, 0, b, off, n);
+ return n;
+ }
+ }
+}
