(camel) branch main updated: CAMEL-25733: camel-mail - remove the idempotent key of the mail whose processExchange throws, not only of the rest of the batch (#29289)

[davsclaus.apache.org via commits]

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 393c2d6a2953 CAMEL-25733: camel-mail - remove the idempotent key of 
the mail whose processExchange throws, not only of the rest of the batch 
(#29289)
393c2d6a2953 is described below

commit 393c2d6a2953ff07b95859d00b01fdd906f639d2
Author: allthingssecurity <49463903+allthingssecur...@users.noreply.github.com>
AuthorDate: Fri Nov 6 13:06:50 2026 +0530

    CAMEL-25733: camel-mail - remove the idempotent key of the mail whose 
processExchange throws, not only of the rest of the batch (#29289)
    
    When processExchange throws for a mail, its idempotent key is now removed 
too, not only the keys of the rest of the batch, so the next poll consumes that 
mail again.
    
    Co-Authored-By: Claude Opus 5.5 <nore...@anthropic.com>
---
 .../apache/camel/component/mail/MailConsumer.java  | 10 +++++-
 .../MailIdempotentRepositoryPollErrorTest.java     | 37 +++++++++++++++++++---
 2 files changed, 42 insertions(+), 5 deletions(-)

diff --git 
a/components/camel-mail/src/main/java/org/apache/camel/component/mail/MailConsumer.java
 
b/components/camel-mail/src/main/java/org/apache/camel/component/mail/MailConsumer.java
index 5e591d791ed1..61c7ccb5b557 100644
--- 
a/components/camel-mail/src/main/java/org/apache/camel/component/mail/MailConsumer.java
+++ 
b/components/camel-mail/src/main/java/org/apache/camel/component/mail/MailConsumer.java
@@ -304,7 +304,15 @@ public class MailConsumer extends 
ScheduledBatchPollingConsumer {
             });
 
             // process the exchange
-            processExchange(exchange);
+            try {
+                processExchange(exchange);
+            } catch (Exception e) {
+                // this exchange is already taken from the batch, so 
processBatch does not give it back: when it was
+                // neither committed nor rolled back (both take the key from 
the exchange), remove its key too, so its
+                // mail is not skipped as a duplicate by a later poll
+                removeIdempotentKey((String) 
exchange.removeProperty(MAIL_MESSAGE_UID));
+                throw e;
+            }
         }
     }
 
diff --git 
a/components/camel-mail/src/test/java/org/apache/camel/component/mail/MailIdempotentRepositoryPollErrorTest.java
 
b/components/camel-mail/src/test/java/org/apache/camel/component/mail/MailIdempotentRepositoryPollErrorTest.java
index d779cc0036fa..e5dde75ed5f7 100644
--- 
a/components/camel-mail/src/test/java/org/apache/camel/component/mail/MailIdempotentRepositoryPollErrorTest.java
+++ 
b/components/camel-mail/src/test/java/org/apache/camel/component/mail/MailIdempotentRepositoryPollErrorTest.java
@@ -78,8 +78,9 @@ class MailIdempotentRepositoryPollErrorTest extends 
CamelTestSupport {
         assertEquals(1, pollErrors.size());
         assertInstanceOf(RejectedExecutionException.class, pollErrors.get(0));
         assertEquals(2, routed.size());
-        // the keys of the two mails that went into routing stay, the keys of 
the three mails after them are removed
-        assertEquals(new TreeSet<>(routed), repo.keys());
+        // the key of the first mail stays (its exchange went through 
processExchange); the key of the mail whose
+        // processExchange threw and the keys of the three mails after it are 
removed
+        assertEquals(new TreeSet<>(routed.subList(0, 1)), repo.keys());
     }
 
     @Test
@@ -102,8 +103,36 @@ class MailIdempotentRepositoryPollErrorTest extends 
CamelTestSupport {
         }
 
         assertEquals(2, routed.size());
-        // releasing the first given-back exchange fails, the keys of all 
three mails after the routed ones are removed
-        assertEquals(new TreeSet<>(routed), repo.keys());
+        // releasing the first given-back exchange fails, the keys of the mail 
whose processExchange threw and of all
+        // three mails after it are removed
+        assertEquals(new TreeSet<>(routed.subList(0, 1)), repo.keys());
+    }
+
+    @Test
+    void mailWhoseProcessExchangeThrowsIsConsumedByTheNextPoll() throws 
Exception {
+        appendMessages(1);
+
+        AtomicBoolean fail = new AtomicBoolean(true);
+        List<String> routed = new ArrayList<>();
+        MailConsumer consumer = createConsumer(exchange -> {
+            routed.add(exchange.getProperty(MailConsumer.MAIL_MESSAGE_UID, 
String.class));
+            if (fail.compareAndSet(true, false)) {
+                throw new RejectedExecutionException("Simulated: cannot route 
the exchange");
+            }
+        });
+        try {
+            consumer.poll();
+            assertEquals(1, pollErrors.size());
+            assertEquals(new TreeSet<>(), repo.keys(), "The key of the mail 
whose processExchange threw must be removed");
+
+            // the mail is still in the folder and is not skipped as a 
duplicate
+            consumer.poll();
+        } finally {
+            consumer.stop();
+        }
+
+        assertEquals(2, routed.size());
+        assertEquals(routed.get(0), routed.get(1));
     }
 
     @Test