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 f3c12e7d79fb CAMEL-24090: camel-file - Drain eagerly-added idempotent
keys on poll exception
f3c12e7d79fb is described below
commit f3c12e7d79fba9e4eb9c9873478b7d7b7a68a9cc
Author: Claus Ibsen <[email protected]>
AuthorDate: Thu Jul 16 07:19:00 2026 +0200
CAMEL-24090: camel-file - Drain eagerly-added idempotent keys on poll
exception
With idempotentEager=true (the default), keys are added to the
idempotent repository during directory scanning. If pollDirectory()
throws mid-scan, the List overload of removeExcessiveInProgressFiles
only removes in-progress keys — eagerly-added idempotent keys leak,
causing files to be permanently skipped on subsequent polls.
CAMEL-21830 fixed the Deque overload used for batch draining but missed
the List overload on the poll-exception path. Mirror the same cleanup:
when idempotentEager is enabled, also call removeExcessiveIdempotentFile
for each file during the List drain.
Closes #24745
Co-Authored-By: Claude Opus 4.6 <[email protected]>
Signed-off-by: Claus Ibsen <[email protected]>
---
.../camel/component/file/GenericFileConsumer.java | 3 +
.../file/FileIdempotentEagerPollExceptionTest.java | 85 ++++++++++++++++++++++
2 files changed, 88 insertions(+)
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..e02c4a5091d8 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
@@ -311,6 +311,9 @@ 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) {
+ removeExcessiveIdempotentFile(file, null);
+ }
}
}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/component/file/FileIdempotentEagerPollExceptionTest.java
b/core/camel-core/src/test/java/org/apache/camel/component/file/FileIdempotentEagerPollExceptionTest.java
new file mode 100644
index 000000000000..b0bddb2a99ee
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/component/file/FileIdempotentEagerPollExceptionTest.java
@@ -0,0 +1,85 @@
+/*
+ * 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.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+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.junit.jupiter.api.Test;
+
+/**
+ * Tests that eagerly-added idempotent keys are drained when pollDirectory
throws mid-scan (CAMEL-24090).
+ *
+ * Scenario: two files exist (a.txt, b.txt); preSort=name guarantees a.txt is
scanned first. A filter throws on b.txt
+ * during the first poll only. Without the fix, a.txt's eagerly-added
idempotent key leaks and the file is never
+ * consumed on subsequent polls.
+ */
+public class FileIdempotentEagerPollExceptionTest extends ContextTestSupport {
+
+ private final AtomicBoolean shouldThrow = new AtomicBoolean(true);
+
+ @Override
+ protected Registry createCamelRegistry() throws Exception {
+ Registry jndi = super.createCamelRegistry();
+ jndi.bind("throwOnceFilter", new ThrowOnceFilter());
+ return jndi;
+ }
+
+ @Test
+ public void testIdempotentKeysDrainedOnPollException() throws Exception {
+ template.sendBodyAndHeader(fileUri(), "aaa", Exchange.FILE_NAME,
"a.txt");
+ template.sendBodyAndHeader(fileUri(), "bbb", Exchange.FILE_NAME,
"b.txt");
+
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedMinimumMessageCount(2);
+
+ context.getRouteController().startRoute("testRoute");
+
+ MockEndpoint.assertIsSatisfied(context, 10, TimeUnit.SECONDS);
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from(fileUri("?noop=true&initialDelay=0&delay=100&preSort=name&filter=#throwOnceFilter"))
+ .routeId("testRoute").noAutoStartup()
+ .convertBodyTo(String.class)
+ .to("mock:result");
+ }
+ };
+ }
+
+ private class ThrowOnceFilter implements GenericFileFilter<Object> {
+ @Override
+ public boolean accept(GenericFile<Object> file) {
+ if (file.isDirectory()) {
+ return true;
+ }
+ if ("b.txt".equals(file.getFileName()) &&
shouldThrow.compareAndSet(true, false)) {
+ throw new RuntimeException("Simulated mid-scan failure");
+ }
+ return true;
+ }
+ }
+}
