This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch backport/25726-to-camel-4.22.x in repository https://gitbox.apache.org/repos/asf/camel.git
commit 8176dcb1cefadc16e4a14b236694bbfe78bea8c7 Author: Claus Ibsen <[email protected]> AuthorDate: Wed Aug 26 18:11:08 2026 +0200 CAMEL-24484: camel-file - readLockRemoveOnCommit=false must not remove a pre-existing idempotent entry Fixes a regression where readLockRemoveOnCommit=false (and readLockRemoveOnRollback=false) was silently ignored: a previously committed idempotent-repository entry got removed the next time a file with the same idempotent key was polled. Two places assumed "file did not start processing" implies "we added this idempotent entry ourselves, so it's safe to remove" - false whenever the entry pre-existed. Fixed by tracking lock ownership explicitly via a new Exchange.FILE_LOCK_IDEMPOTENT_ACQUIRED property, and by guarding GenericFileConsumer's "not started" cleanup with isIdempotent(). Co-authored-by: Claude Opus 4.8 <[email protected]> Co-Authored-By: Claude Sonnet 5 <[email protected]> Closes #25726 (cherry picked from commit 1f35d963cd79366d6b102cf1cadb90ce50c711d3) Signed-off-by: Claus Ibsen <[email protected]> --- .../camel/component/file/GenericFileConsumer.java | 6 +- ...dempotentChangedRepositoryReadLockStrategy.java | 25 ++++- ...IdempotentRenameRepositoryReadLockStrategy.java | 25 ++++- .../FileIdempotentRepositoryReadLockStrategy.java | 24 ++++- .../org/apache/camel/ExchangeConstantProvider.java | 3 +- .../src/main/java/org/apache/camel/Exchange.java | 1 + ...dempotentChangedReadLockRemoveOnCommitTest.java | 82 ++++++++++++++++ ...mpotentRepositoryReadLockStrategyAbortTest.java | 105 +++++++++++++++++++++ 8 files changed, 262 insertions(+), 9 deletions(-) diff --git a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java index fe72a23124b1..f42276871293 100644 --- a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java +++ b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileConsumer.java @@ -284,7 +284,8 @@ public abstract class GenericFileConsumer<T> extends ScheduledBatchPollingConsum String key = file.getAbsoluteFilePath(); endpoint.getInProgressRepository().remove(key); // if we added eager to idempotent then we need to remove this - if (endpoint.isIdempotentEager() && endpoint.getIdempotentRepository() != null) { + if (Boolean.TRUE.equals(endpoint.isIdempotent()) && endpoint.isIdempotentEager() + && endpoint.getIdempotentRepository() != null) { removeExcessiveIdempotentFile(file, null); } releaseExchange(exchange, true); @@ -311,7 +312,8 @@ public abstract class GenericFileConsumer<T> extends ScheduledBatchPollingConsum for (GenericFile file : files) { String key = file.getAbsoluteFilePath(); endpoint.getInProgressRepository().remove(key); - if (endpoint.isIdempotentEager() && endpoint.getIdempotentRepository() != null) { + if (Boolean.TRUE.equals(endpoint.isIdempotent()) && endpoint.isIdempotentEager() + && endpoint.getIdempotentRepository() != null) { removeExcessiveIdempotentFile(file, null); } } diff --git a/components/camel-file/src/main/java/org/apache/camel/component/file/strategy/FileIdempotentChangedRepositoryReadLockStrategy.java b/components/camel-file/src/main/java/org/apache/camel/component/file/strategy/FileIdempotentChangedRepositoryReadLockStrategy.java index 945277336efc..ce5e6c8c355a 100644 --- a/components/camel-file/src/main/java/org/apache/camel/component/file/strategy/FileIdempotentChangedRepositoryReadLockStrategy.java +++ b/components/camel-file/src/main/java/org/apache/camel/component/file/strategy/FileIdempotentChangedRepositoryReadLockStrategy.java @@ -108,6 +108,10 @@ public class FileIdempotentChangedRepositoryReadLockStrategy extends ServiceSupp idempotentRepository.remove(exchange, key); } } + // remember whether we own the key end-to-end (idempotent add + changed check both + // succeeded), so releaseExclusiveReadLockOnAbort knows whether it is safe to remove it if + // begin() later fails for an unrelated reason + exchange.setProperty(asReadLockKey(file, Exchange.FILE_LOCK_IDEMPOTENT_ACQUIRED), answer); return answer; } @@ -115,8 +119,16 @@ public class FileIdempotentChangedRepositoryReadLockStrategy extends ServiceSupp public void releaseExclusiveReadLockOnAbort( GenericFileOperations<File> operations, GenericFile<File> file, Exchange exchange) throws Exception { - String key = asKey(exchange, file); - idempotentRepository.remove(exchange, key); + // only remove the key if we own it end-to-end. If we never owned it (pre-existing key) or + // the changed-check failed (already cleaned up above in acquireExclusiveReadLock) there is + // nothing to do here. If we do own it, acquireExclusiveReadLock succeeded and something + // later in begin() must have failed, so the key is ours to clean up + boolean acquired + = exchange.getProperty(asReadLockKey(file, Exchange.FILE_LOCK_IDEMPOTENT_ACQUIRED), false, Boolean.class); + if (acquired) { + String key = asKey(exchange, file); + idempotentRepository.remove(exchange, key); + } changed.releaseExclusiveReadLockOnAbort(operations, file, exchange); } @@ -328,6 +340,15 @@ public class FileIdempotentChangedRepositoryReadLockStrategy extends ServiceSupp return key; } + private static String asReadLockKey(GenericFile<File> file, String key) { + // use the copy from absolute path as that was the original path of the + // file when the lock was acquired; e.g. if the file consumer uses preMove + // then the file is moved and would otherwise no longer match + String path = file.getCopyFromAbsoluteFilePath() != null + ? file.getCopyFromAbsoluteFilePath() : file.getAbsoluteFilePath(); + return path + "-" + key; + } + @Override protected void doStart() throws Exception { ObjectHelper.notNull(camelContext, "camelContext", this); diff --git a/components/camel-file/src/main/java/org/apache/camel/component/file/strategy/FileIdempotentRenameRepositoryReadLockStrategy.java b/components/camel-file/src/main/java/org/apache/camel/component/file/strategy/FileIdempotentRenameRepositoryReadLockStrategy.java index 72ca3c2a866a..0f205dc20bf8 100644 --- a/components/camel-file/src/main/java/org/apache/camel/component/file/strategy/FileIdempotentRenameRepositoryReadLockStrategy.java +++ b/components/camel-file/src/main/java/org/apache/camel/component/file/strategy/FileIdempotentRenameRepositoryReadLockStrategy.java @@ -100,6 +100,10 @@ public class FileIdempotentRenameRepositoryReadLockStrategy extends ServiceSuppo idempotentRepository.remove(exchange, key); } } + // remember whether we own the key end-to-end (idempotent add + rename check both + // succeeded), so releaseExclusiveReadLockOnAbort knows whether it is safe to remove it if + // begin() later fails for an unrelated reason + exchange.setProperty(asReadLockKey(file, Exchange.FILE_LOCK_IDEMPOTENT_ACQUIRED), answer); return answer; } @@ -107,8 +111,16 @@ public class FileIdempotentRenameRepositoryReadLockStrategy extends ServiceSuppo public void releaseExclusiveReadLockOnAbort( GenericFileOperations<File> operations, GenericFile<File> file, Exchange exchange) throws Exception { - String key = asKey(exchange, file); - idempotentRepository.remove(exchange, key); + // only remove the key if we own it end-to-end. If we never owned it (pre-existing key) or + // the rename-check failed (already cleaned up above in acquireExclusiveReadLock) there is + // nothing to do here. If we do own it, acquireExclusiveReadLock succeeded and something + // later in begin() must have failed, so the key is ours to clean up + boolean acquired + = exchange.getProperty(asReadLockKey(file, Exchange.FILE_LOCK_IDEMPOTENT_ACQUIRED), false, Boolean.class); + if (acquired) { + String key = asKey(exchange, file); + idempotentRepository.remove(exchange, key); + } rename.releaseExclusiveReadLockOnAbort(operations, file, exchange); } @@ -242,6 +254,15 @@ public class FileIdempotentRenameRepositoryReadLockStrategy extends ServiceSuppo return key; } + private static String asReadLockKey(GenericFile<File> file, String key) { + // use the copy from absolute path as that was the original path of the + // file when the lock was acquired; e.g. if the file consumer uses preMove + // then the file is moved and would otherwise no longer match + String path = file.getCopyFromAbsoluteFilePath() != null + ? file.getCopyFromAbsoluteFilePath() : file.getAbsoluteFilePath(); + return path + "-" + key; + } + @Override protected void doStart() throws Exception { ObjectHelper.notNull(camelContext, "camelContext", this); diff --git a/components/camel-file/src/main/java/org/apache/camel/component/file/strategy/FileIdempotentRepositoryReadLockStrategy.java b/components/camel-file/src/main/java/org/apache/camel/component/file/strategy/FileIdempotentRepositoryReadLockStrategy.java index 94c124619af2..1883051b947d 100644 --- a/components/camel-file/src/main/java/org/apache/camel/component/file/strategy/FileIdempotentRepositoryReadLockStrategy.java +++ b/components/camel-file/src/main/java/org/apache/camel/component/file/strategy/FileIdempotentRepositoryReadLockStrategy.java @@ -88,6 +88,9 @@ public class FileIdempotentRepositoryReadLockStrategy extends ServiceSupport // another node is processing the file so skip CamelLogger.log(LOG, readLockLoggingLevel, "Cannot acquire read lock. Will skip the file: " + file); } + // remember whether we added the key ourselves, so releaseExclusiveReadLockOnAbort knows + // whether it is safe to remove it if begin() later fails for an unrelated reason + exchange.setProperty(asReadLockKey(file, Exchange.FILE_LOCK_IDEMPOTENT_ACQUIRED), answer); return answer; } @@ -95,8 +98,16 @@ public class FileIdempotentRepositoryReadLockStrategy extends ServiceSupport public void releaseExclusiveReadLockOnAbort( GenericFileOperations<File> operations, GenericFile<File> file, Exchange exchange) throws Exception { - String key = asKey(exchange, file); - idempotentRepository.remove(exchange, key); + // only remove the key if we added it during acquireExclusiveReadLock. If the key already + // existed (owned by a previous run or another node) it must not be touched here. If we did + // add it, acquireExclusiveReadLock succeeded and something later in begin() must have + // failed, so the key is ours to clean up + boolean acquired + = exchange.getProperty(asReadLockKey(file, Exchange.FILE_LOCK_IDEMPOTENT_ACQUIRED), false, Boolean.class); + if (acquired) { + String key = asKey(exchange, file); + idempotentRepository.remove(exchange, key); + } } @Override @@ -290,6 +301,15 @@ public class FileIdempotentRepositoryReadLockStrategy extends ServiceSupport return key; } + private static String asReadLockKey(GenericFile<File> file, String key) { + // use the copy from absolute path as that was the original path of the + // file when the lock was acquired; e.g. if the file consumer uses preMove + // then the file is moved and would otherwise no longer match + String path = file.getCopyFromAbsoluteFilePath() != null + ? file.getCopyFromAbsoluteFilePath() : file.getAbsoluteFilePath(); + return path + "-" + key; + } + @Override protected void doStart() throws Exception { ObjectHelper.notNull(camelContext, "camelContext", this); diff --git a/core/camel-api/src/generated/java/org/apache/camel/ExchangeConstantProvider.java b/core/camel-api/src/generated/java/org/apache/camel/ExchangeConstantProvider.java index ec7e00bb731b..9b93e7bb3081 100644 --- a/core/camel-api/src/generated/java/org/apache/camel/ExchangeConstantProvider.java +++ b/core/camel-api/src/generated/java/org/apache/camel/ExchangeConstantProvider.java @@ -17,7 +17,7 @@ public class ExchangeConstantProvider { private static final Map<String, String> MAP; static { - Map<String, String> map = new HashMap<>(155); + Map<String, String> map = new HashMap<>(156); map.put("ACTIVITY_SPAN_TAGS", "CamelActivitySpanTags"); map.put("AGGREGATED_COLLECTION_GUARD", "CamelAggregatedCollectionGuard"); map.put("AGGREGATED_COMPLETED_BY", "CamelAggregatedCompletedBy"); @@ -73,6 +73,7 @@ public class ExchangeConstantProvider { map.put("FILE_LOCK_EXCLUSIVE_LOCK", "CamelFileLockExclusiveLock"); map.put("FILE_LOCK_FILE_ACQUIRED", "CamelFileLockFileAcquired"); map.put("FILE_LOCK_FILE_NAME", "CamelFileLockFileName"); + map.put("FILE_LOCK_IDEMPOTENT_ACQUIRED", "CamelFileLockIdempotentAcquired"); map.put("FILE_LOCK_RANDOM_ACCESS_FILE", "CamelFileLockRandomAccessFile"); map.put("FILE_NAME", "CamelFileName"); map.put("FILE_NAME_CONSUMED", "CamelFileNameConsumed"); diff --git a/core/camel-api/src/main/java/org/apache/camel/Exchange.java b/core/camel-api/src/main/java/org/apache/camel/Exchange.java index d2b64260536d..74457a08e0e8 100644 --- a/core/camel-api/src/main/java/org/apache/camel/Exchange.java +++ b/core/camel-api/src/main/java/org/apache/camel/Exchange.java @@ -156,6 +156,7 @@ public interface Exchange extends VariableAware { String FILE_LENGTH = "CamelFileLength"; String FILE_LOCK_FILE_ACQUIRED = "CamelFileLockFileAcquired"; String FILE_LOCK_FILE_NAME = "CamelFileLockFileName"; + String FILE_LOCK_IDEMPOTENT_ACQUIRED = "CamelFileLockIdempotentAcquired"; String FILE_LOCK_EXCLUSIVE_LOCK = "CamelFileLockExclusiveLock"; String FILE_LOCK_RANDOM_ACCESS_FILE = "CamelFileLockRandomAccessFile"; String FILE_LOCK_CHANNEL_FILE = "CamelFileLockChannelFile"; diff --git a/core/camel-core/src/test/java/org/apache/camel/component/file/strategy/FileIdempotentChangedReadLockRemoveOnCommitTest.java b/core/camel-core/src/test/java/org/apache/camel/component/file/strategy/FileIdempotentChangedReadLockRemoveOnCommitTest.java new file mode 100644 index 000000000000..1d35ca58c1cf --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/component/file/strategy/FileIdempotentChangedReadLockRemoveOnCommitTest.java @@ -0,0 +1,82 @@ +/* + * 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.strategy; + +import java.util.concurrent.TimeUnit; + +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Exchange; +import org.apache.camel.builder.RouteBuilder; +import org.apache.camel.component.mock.MockEndpoint; +import org.apache.camel.spi.Registry; +import org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; + +class FileIdempotentChangedReadLockRemoveOnCommitTest extends ContextTestSupport { + + final MemoryIdempotentRepository myRepo = new MemoryIdempotentRepository(); + + @Override + protected Registry createCamelRegistry() throws Exception { + Registry jndi = super.createCamelRegistry(); + jndi.bind("myRepo", myRepo); + return jndi; + } + + @Test + void testExistingEntryNotRemovedOnCommitFalse() throws Exception { + assertThat(myRepo.getCacheSize()).isZero(); + + MockEndpoint mock = getMockEndpoint("mock:result"); + mock.expectedMessageCount(1); + + // drop hello.txt -> processed once, moved to .camel, entry retained + template.sendBodyAndHeader(fileUri(), "Hello World", Exchange.FILE_NAME, "hello.txt"); + mock.assertIsSatisfied(); + + await().atMost(5, TimeUnit.SECONDS).untilAsserted(() -> assertThat(myRepo.contains("hello.txt")).isTrue()); + + // re-drop a file with the same name -> must be skipped as a duplicate + // and the existing entry must be retained because readLockRemoveOnCommit=false + mock.reset(); + mock.expectedMessageCount(0); + template.sendBodyAndHeader(fileUri(), "Hello World Again", Exchange.FILE_NAME, "hello.txt"); + + // give the consumer several poll cycles to (wrongly) process/remove it + mock.assertIsSatisfied(2000); + + assertThat(myRepo.contains("hello.txt")) + .as("existing idempotent entry must be retained when readLockRemoveOnCommit=false") + .isTrue(); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + @Override + public void configure() { + from(fileUri("?initialDelay=0&delay=50&readLockCheckInterval=50" + + "&readLock=idempotent-changed&idempotentRepository=#myRepo" + + "&idempotentKey=${file:onlyname}&readLockRemoveOnCommit=false")) + .to("mock:result"); + } + }; + } +} diff --git a/core/camel-core/src/test/java/org/apache/camel/component/file/strategy/FileIdempotentRepositoryReadLockStrategyAbortTest.java b/core/camel-core/src/test/java/org/apache/camel/component/file/strategy/FileIdempotentRepositoryReadLockStrategyAbortTest.java new file mode 100644 index 000000000000..adeb287c4ada --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/component/file/strategy/FileIdempotentRepositoryReadLockStrategyAbortTest.java @@ -0,0 +1,105 @@ +/* + * 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.strategy; + +import java.io.File; +import java.nio.file.Files; + +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Exchange; +import org.apache.camel.component.file.FileEndpoint; +import org.apache.camel.component.file.GenericFile; +import org.apache.camel.support.DefaultExchange; +import org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository; +import org.apache.camel.support.service.ServiceHelper; +import org.junit.jupiter.api.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Unit tests releaseExclusiveReadLockOnAbort directly: it must only remove the idempotent key when this strategy itself + * added it during acquireExclusiveReadLock. abort() can also be reached after a successful acquire, e.g. when a preMove + * rename fails afterwards, so a pre-existing key (owned by a previous run or another node) must never be touched, while + * a key we own must be cleaned up. + */ +class FileIdempotentRepositoryReadLockStrategyAbortTest extends ContextTestSupport { + + private FileIdempotentRepositoryReadLockStrategy newStrategy(MemoryIdempotentRepository repo) throws Exception { + ServiceHelper.startService(repo); + + FileEndpoint endpoint = context.getEndpoint(fileUri(), FileEndpoint.class); + + FileIdempotentRepositoryReadLockStrategy strategy = new FileIdempotentRepositoryReadLockStrategy(); + strategy.setCamelContext(context); + strategy.setIdempotentRepository(repo); + strategy.prepareOnStartup(null, endpoint); + ServiceHelper.startService(strategy); + return strategy; + } + + private GenericFile<File> newGenericFile(String name) throws Exception { + File file = testFile(name).toFile(); + Files.createDirectories(file.getParentFile().toPath()); + Files.writeString(file.toPath(), "Hello World"); + + GenericFile<File> genericFile = new GenericFile<>(); + genericFile.setFile(file); + genericFile.setAbsoluteFilePath(file.getAbsolutePath()); + return genericFile; + } + + @Test + void testAbortDoesNotRemovePreExistingKey() throws Exception { + MemoryIdempotentRepository repo = new MemoryIdempotentRepository(); + FileIdempotentRepositoryReadLockStrategy strategy = newStrategy(repo); + + GenericFile<File> genericFile = newGenericFile("hello.txt"); + Exchange exchange = new DefaultExchange(context); + + // simulate the key already being owned by a previous, already-committed run + repo.add(exchange, genericFile.getAbsoluteFilePath()); + + boolean acquired = strategy.acquireExclusiveReadLock(null, genericFile, exchange); + assertThat(acquired).as("must not acquire a key that already exists").isFalse(); + + strategy.releaseExclusiveReadLockOnAbort(null, genericFile, exchange); + + assertThat(repo.contains(genericFile.getAbsoluteFilePath())) + .as("pre-existing key must be retained since we never owned it") + .isTrue(); + } + + @Test + void testAbortRemovesKeyWeAcquired() throws Exception { + MemoryIdempotentRepository repo = new MemoryIdempotentRepository(); + FileIdempotentRepositoryReadLockStrategy strategy = newStrategy(repo); + + GenericFile<File> genericFile = newGenericFile("hello2.txt"); + Exchange exchange = new DefaultExchange(context); + + boolean acquired = strategy.acquireExclusiveReadLock(null, genericFile, exchange); + assertThat(acquired).as("must acquire a key that does not yet exist").isTrue(); + + // simulate begin() throwing afterwards for an unrelated reason (e.g. a preMove rename + // failing), which still routes into releaseExclusiveReadLockOnAbort + strategy.releaseExclusiveReadLockOnAbort(null, genericFile, exchange); + + assertThat(repo.contains(genericFile.getAbsoluteFilePath())) + .as("key we acquired ourselves must be removed on abort") + .isFalse(); + } +}
