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 52c2942c0653 CAMEL-24091: camel-file - Snapshot idempotent key at poll
time to prevent key mismatch (#24746)
52c2942c0653 is described below
commit 52c2942c06531fcde2048e68c171c012102783ef
Author: Claus Ibsen <[email protected]>
AuthorDate: Thu Jul 16 09:49:16 2026 +0200
CAMEL-24091: camel-file - Snapshot idempotent key at poll time to prevent
key mismatch (#24746)
CAMEL-24091: camel-file - Snapshot idempotent key at poll time to prevent
key mismatch
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 e02c4a5091d8..76a775a91b74 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
@@ -410,6 +410,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");
+ }
+ };
+ }
+}
