This is an automated email from the ASF dual-hosted git repository. davsclaus pushed a commit to branch fix/CAMEL-24091 in repository https://gitbox.apache.org/repos/asf/camel.git
commit f0ccead6312a092789b8a7aad0142f8631e31b43 Author: Claus Ibsen <[email protected]> AuthorDate: Wed Jul 15 20:36:22 2026 +0200 CAMEL-24091: camel-file - Snapshot idempotent key at poll time to prevent key mismatch when route mutates file headers Co-Authored-By: Claude Opus 4.6 <[email protected]> Signed-off-by: Claus Ibsen <[email protected]> --- .../camel/component/file/GenericFileConsumer.java | 7 +++ .../component/file/GenericFileOnCompletion.java | 12 ++-- .../org/apache/camel/ExchangeConstantProvider.java | 3 +- .../src/main/java/org/apache/camel/Exchange.java | 1 + ...ileConsumerIdempotentKeyHeaderMutationTest.java | 66 ++++++++++++++++++++++ 5 files changed, 80 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 89b676b33a82..6287496b77f6 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 @@ -407,6 +407,13 @@ public abstract class GenericFileConsumer<T> extends ScheduledBatchPollingConsum // and then the file name would be changed String absoluteFileName = file.getAbsoluteFilePath(); + // snapshot the idempotent key now while the exchange still has pristine file headers; + // after routing, headers like CamelFileName may be mutated, causing key mismatch at completion time + if (Boolean.TRUE.equals(endpoint.isIdempotent()) && endpoint.getIdempotentKey() != null) { + String key = endpoint.getIdempotentKey().evaluate(exchange, String.class); + exchange.setProperty(Exchange.FILE_IDEMPOTENT_KEY, key); + } + // check if we can begin processing the file Exception beginCause = null; boolean begin = false; diff --git a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileOnCompletion.java b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileOnCompletion.java index f731aff6faf7..9f6cfbad39e2 100644 --- a/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileOnCompletion.java +++ b/components/camel-file/src/main/java/org/apache/camel/component/file/GenericFileOnCompletion.java @@ -112,12 +112,10 @@ public class GenericFileOnCompletion<T> implements Synchronization { protected void processStrategyCommit( GenericFileProcessStrategy<T> processStrategy, Exchange exchange, GenericFile<T> file) { if (Boolean.TRUE.equals(endpoint.isIdempotent())) { - // use absolute file path as default key, but evaluate if an - // expression key was configured + // use the poll-time snapshot so header mutations during routing don't cause key mismatch String key = absoluteFileName; if (endpoint.getIdempotentKey() != null) { - Exchange dummy = GenericFileHelper.createDummy(endpoint, exchange, () -> file); - key = endpoint.getIdempotentKey().evaluate(dummy, String.class); + key = exchange.getProperty(Exchange.FILE_IDEMPOTENT_KEY, absoluteFileName, String.class); } // only add to idempotent repository if we could process the file if (key != null) { @@ -155,12 +153,10 @@ public class GenericFileOnCompletion<T> implements Synchronization { } if (Boolean.TRUE.equals(endpoint.isIdempotent())) { - // use absolute file path as default key, but evaluate if an - // expression key was configured + // use the poll-time snapshot so header mutations during routing don't cause key mismatch String key = absoluteFileName; if (endpoint.getIdempotentKey() != null) { - Exchange dummy = GenericFileHelper.createDummy(endpoint, exchange, () -> file); - key = endpoint.getIdempotentKey().evaluate(dummy, String.class); + key = exchange.getProperty(Exchange.FILE_IDEMPOTENT_KEY, absoluteFileName, String.class); } if (key != null) { endpoint.getIdempotentRepository().remove(key); 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 6522152b19f2..ec7e00bb731b 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<>(154); + Map<String, String> map = new HashMap<>(155); map.put("ACTIVITY_SPAN_TAGS", "CamelActivitySpanTags"); map.put("AGGREGATED_COLLECTION_GUARD", "CamelAggregatedCollectionGuard"); map.put("AGGREGATED_COMPLETED_BY", "CamelAggregatedCompletedBy"); @@ -65,6 +65,7 @@ public class ExchangeConstantProvider { map.put("FATAL_FALLBACK_ERROR_HANDLER", "CamelFatalFallbackErrorHandler"); map.put("FILE_CONTENT_TYPE", "CamelFileContentType"); map.put("FILE_EXCHANGE_FILE", "CamelFileExchangeFile"); + map.put("FILE_IDEMPOTENT_KEY", "CamelFileIdempotentKey"); map.put("FILE_LAST_MODIFIED", "CamelFileLastModified"); map.put("FILE_LENGTH", "CamelFileLength"); map.put("FILE_LOCAL_WORK_PATH", "CamelFileLocalWorkPath"); 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 c764291302a2..d2b64260536d 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 @@ -160,6 +160,7 @@ public interface Exchange extends VariableAware { String FILE_LOCK_RANDOM_ACCESS_FILE = "CamelFileLockRandomAccessFile"; String FILE_LOCK_CHANNEL_FILE = "CamelFileLockChannelFile"; String FILE_EXCHANGE_FILE = "CamelFileExchangeFile"; + String FILE_IDEMPOTENT_KEY = "CamelFileIdempotentKey"; String FILTER_NON_XML_CHARS = "CamelFilterNonXmlChars"; String GROUPED_EXCHANGE = "CamelGroupedExchange"; diff --git a/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerIdempotentKeyHeaderMutationTest.java b/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerIdempotentKeyHeaderMutationTest.java new file mode 100644 index 000000000000..6365e849e695 --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerIdempotentKeyHeaderMutationTest.java @@ -0,0 +1,66 @@ +/* + * 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.util.UUID; +import java.util.concurrent.atomic.AtomicInteger; + +import org.apache.camel.ContextTestSupport; +import org.apache.camel.Exchange; +import org.apache.camel.Processor; +import org.apache.camel.builder.RouteBuilder; +import org.junit.jupiter.api.Test; + +/** + * Test that mutating CamelFileName in a route does not break idempotent key evaluation at completion time. Before the + * fix, rollback would remove the wrong key (computed from mutated headers), leaving the eagerly-added poll-time key + * orphaned so the file was never retried. + */ +public class FileConsumerIdempotentKeyHeaderMutationTest extends ContextTestSupport { + + private static final String TEST_FILE_NAME = "report-" + UUID.randomUUID() + ".txt"; + private final AtomicInteger attempts = new AtomicInteger(); + + @Test + public void testIdempotentKeyUnaffectedByHeaderMutation() throws Exception { + getMockEndpoint("mock:result").expectedBodiesReceived("Hello World"); + + template.sendBodyAndHeader(fileUri(), "Hello World", Exchange.FILE_NAME, TEST_FILE_NAME); + + assertMockEndpointsSatisfied(); + } + + @Override + protected RouteBuilder createRouteBuilder() { + return new RouteBuilder() { + public void configure() { + from(fileUri("?idempotent=true&idempotentKey=${file:onlyname}-${file:size}" + + "&initialDelay=0&delay=10")) + .setHeader(Exchange.FILE_NAME, constant("output.txt")) + .process(new Processor() { + @Override + public void process(Exchange exchange) throws Exception { + if (attempts.incrementAndGet() == 1) { + throw new IllegalArgumentException("Simulated failure on first attempt"); + } + } + }) + .to("mock:result"); + } + }; + } +}
