(camel) branch fc created (now 953e8616136)

[da...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

davsclaus pushed a change to branch fc
in repository https://gitbox.apache.org/repos/asf/camel.git


      at 953e8616136 CAMEL-21830: camel-file - Using consumer template to consume a single file issue with idempotentEager

This branch includes the following new commits:

     new 953e8616136 CAMEL-21830: camel-file - Using consumer template to consume a single file issue with idempotentEager

The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails.  The revisions
listed as "add" were already present in the repository and have only
been added to this reference.

(camel) 01/01: CAMEL-21830: camel-file - Using consumer template to consume a single file issue with idempotentEager

[da...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

davsclaus pushed a commit to branch fc
in repository https://gitbox.apache.org/repos/asf/camel.git

commit 953e86161365c271f3db66e2da45a821128bc351
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Tue Mar 4 16:10:21 2025 +0100

    CAMEL-21830: camel-file - Using consumer template to consume a single file issue with idempotentEager
---
 .../camel/component/file/GenericFileConsumer.java  | 17 +++++-
 .../FileConsumerTemplateIdempotentEagerTest.java   | 60 ++++++++++++++++++++++
 2 files changed, 76 insertions(+), 1 deletion(-)

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 73c0c3c82d9..d4bece9f506 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
@@ -274,13 +274,28 @@ public abstract class GenericFileConsumer<T> extends ScheduledBatchPollingConsum
         while (exchanges.size() > limit) {
             // must remove last
             Exchange exchange = exchanges.removeLast();
-            GenericFile<?> file = exchange.getProperty(ExchangePropertyKey.FILE_EXCHANGE_FILE, GenericFile.class);
+            GenericFile file = exchange.getProperty(ExchangePropertyKey.FILE_EXCHANGE_FILE, GenericFile.class);
             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) {
+                removeExcessiveIdempotentFile(file, null);
+            }
             releaseExchange(exchange, true);
         }
     }
 
+    void removeExcessiveIdempotentFile(GenericFile file, Exchange dynamic) {
+        String key = file.getAbsoluteFilePath();
+        if (endpoint.getIdempotentKey() != null) {
+            Exchange dummy = GenericFileHelper.createDummy(endpoint, dynamic, () -> file);
+            key = endpoint.getIdempotentKey().evaluate(dummy, String.class);
+        }
+        if (key != null) {
+            endpoint.getIdempotentRepository().remove(key);
+        }
+    }
+
     /**
      * Drain any in progress files as we are done with the files
      *
diff --git a/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerTemplateIdempotentEagerTest.java b/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerTemplateIdempotentEagerTest.java
new file mode 100644
index 00000000000..f6eeff87770
--- /dev/null
+++ b/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumerTemplateIdempotentEagerTest.java
@@ -0,0 +1,60 @@
+/*
+ * 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 org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+public class FileConsumerTemplateIdempotentEagerTest extends ContextTestSupport {
+
+    @Override
+    public boolean isUseRouteBuilder() {
+        return false;
+    }
+
+    @Test
+    public void testConsumerTemplate() throws Exception {
+        context.start();
+
+        String uri = fileUri() + "?noop=true";
+
+        template.sendBodyAndHeader(uri, "one", Exchange.FILE_NAME, "1.txt");
+        template.sendBodyAndHeader(uri, "two", Exchange.FILE_NAME, "2.txt");
+
+        Exchange e1 = consumer().receive(uri, 5000);
+        Assertions.assertNotNull(e1);
+        String b1 = e1.getMessage().getBody(String.class);
+        Assertions.assertTrue("one".equals(b1) || "two".equals(b1), "Should either be one or two, was: " + b1);
+        consumer().doneUoW(e1);
+
+        FileEndpoint fe = context.getEndpoint(uri, FileEndpoint.class);
+        MemoryIdempotentRepository repo = (MemoryIdempotentRepository) fe.getIdempotentRepository();
+        Assertions.assertEquals(1, repo.getCacheSize());
+
+        Exchange e2 = consumer().receive(uri, 5000);
+        Assertions.assertNotNull(e2);
+        String b2 = e2.getMessage().getBody(String.class);
+        Assertions.assertTrue("one".equals(b2) || "two".equals(b2), "Should either be one or two, was: " + b2);
+        consumer().doneUoW(e2);
+        Assertions.assertEquals(2, repo.getCacheSize());
+
+        Assertions.assertNotEquals(b1, b2);
+    }
+}