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 96d9614fe237 CAMEL-26707: camel-core - Idempotent Consumer: remove the
message id of an exchange marked rollback only (#29404)
96d9614fe237 is described below
commit 96d9614fe23761879ec6544090ca8d91a53ca623
Author: allthingssecurity <[email protected]>
AuthorDate: Sun Nov 8 16:31:00 2026 +0530
CAMEL-26707: camel-core - Idempotent Consumer: remove the message id of an
exchange marked rollback only (#29404)
* CAMEL-26707: camel-core - Idempotent Consumer: remove the message id of
an exchange marked rollback only
* CAMEL-26707: camel-core - Idempotent Consumer docs: a rollback only
exchange counts as failed
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/catalog/docs/idempotentConsumer-eip.adoc | 3 +
.../modules/eips/pages/idempotentConsumer-eip.adoc | 3 +
.../idempotent/IdempotentOnCompletion.java | 3 +
.../IdempotentConsumerRollbackOnlyTest.java | 120 +++++++++++++++++++++
.../ROOT/pages/camel-4x-upgrade-guide-4_24.adoc | 13 +++
5 files changed, 142 insertions(+)
diff --git
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/idempotentConsumer-eip.adoc
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/idempotentConsumer-eip.adoc
index d4f2a23cbc95..8510903d15f4 100644
---
a/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/idempotentConsumer-eip.adoc
+++
b/catalog/camel-catalog/src/generated/resources/org/apache/camel/catalog/docs/idempotentConsumer-eip.adoc
@@ -17,6 +17,9 @@ Camel will add the message id eagerly to the repository to
detect
duplication also for xref:manual::exchange.adoc[Exchange]'s' currently in
progress.
On completion Camel will remove the message id from the repository if
the xref:manual::exchange.adoc[Exchange] failed, otherwise it stays there.
+An Exchange marked with `markRollbackOnly()` or `markRollbackOnlyLast()`
counts as
+failed too, even without an exception. Set `removeOnFailure=false` to keep the
message id
+of a failed Exchange.
== Options
diff --git
a/core/camel-core-engine/src/main/docs/modules/eips/pages/idempotentConsumer-eip.adoc
b/core/camel-core-engine/src/main/docs/modules/eips/pages/idempotentConsumer-eip.adoc
index d4f2a23cbc95..8510903d15f4 100644
---
a/core/camel-core-engine/src/main/docs/modules/eips/pages/idempotentConsumer-eip.adoc
+++
b/core/camel-core-engine/src/main/docs/modules/eips/pages/idempotentConsumer-eip.adoc
@@ -17,6 +17,9 @@ Camel will add the message id eagerly to the repository to
detect
duplication also for xref:manual::exchange.adoc[Exchange]'s' currently in
progress.
On completion Camel will remove the message id from the repository if
the xref:manual::exchange.adoc[Exchange] failed, otherwise it stays there.
+An Exchange marked with `markRollbackOnly()` or `markRollbackOnlyLast()`
counts as
+failed too, even without an exception. Set `removeOnFailure=false` to keep the
message id
+of a failed Exchange.
== Options
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/idempotent/IdempotentOnCompletion.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/idempotent/IdempotentOnCompletion.java
index 88c78327b193..22fdde87ff00 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/idempotent/IdempotentOnCompletion.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/idempotent/IdempotentOnCompletion.java
@@ -50,6 +50,9 @@ public class IdempotentOnCompletion implements
Synchronization {
// the exchange did not process successfully but was failure
handled by the dead letter channel
// and thus moved to the dead letter queue. We should thus not
consider it as complete.
onFailedMessage(exchange, messageId);
+ } else if (exchange.isRollbackOnly() || exchange.isRollbackOnlyLast())
{
+ // a rollback only exchange (without an exception) did not
complete successfully either
+ onFailedMessage(exchange, messageId);
} else {
onCompletedMessage(exchange, messageId);
}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/IdempotentConsumerRollbackOnlyTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/IdempotentConsumerRollbackOnlyTest.java
new file mode 100644
index 000000000000..6ad6230e3f80
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/IdempotentConsumerRollbackOnlyTest.java
@@ -0,0 +1,120 @@
+/*
+ * 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.processor;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.spi.IdempotentRepository;
+import
org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * An exchange marked rollback only (without an exception) did not complete
successfully: the Idempotent Consumer must
+ * remove its message id (removeOnFailure=true is the default), so a
redelivered message is not dropped as a duplicate.
+ */
+class IdempotentConsumerRollbackOnlyTest extends ContextTestSupport {
+
+ private final IdempotentRepository repository =
MemoryIdempotentRepository.memoryIdempotentRepository(200);
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("direct:rollbackOnly").idempotentConsumer(header("messageId"), repository)
+ .to("mock:result")
+ .filter(header("rollback")).markRollbackOnly();
+
+
from("direct:rollbackOnlyLast").idempotentConsumer(header("messageId"),
repository)
+ .to("mock:result")
+ .filter(header("rollback")).markRollbackOnlyLast();
+
+
from("direct:completionEager").idempotentConsumer(header("messageId"),
repository).completionEager(true)
+ .to("mock:result")
+ .filter(header("rollback")).markRollbackOnly();
+
+
from("direct:notEager").idempotentConsumer(header("messageId"),
repository).eager(false)
+ .to("mock:result")
+ .filter(header("rollback")).markRollbackOnly();
+
+
from("direct:keepOnFailure").idempotentConsumer(header("messageId"),
repository).removeOnFailure(false)
+ .to("mock:result")
+ .filter(header("rollback")).markRollbackOnly();
+ }
+ };
+ }
+
+ @Test
+ void testRollbackOnlyRemovesMessageId() throws Exception {
+ assertRedeliveryProcessed("direct:rollbackOnly");
+ }
+
+ @Test
+ void testRollbackOnlyLastRemovesMessageId() throws Exception {
+ assertRedeliveryProcessed("direct:rollbackOnlyLast");
+ }
+
+ @Test
+ void testRollbackOnlyCompletionEagerRemovesMessageId() throws Exception {
+ assertRedeliveryProcessed("direct:completionEager");
+ }
+
+ @Test
+ void testRollbackOnlyNotEagerDoesNotAddMessageId() throws Exception {
+ assertRedeliveryProcessed("direct:notEager");
+ }
+
+ @Test
+ void testRollbackOnlyKeepsMessageIdWhenNotRemovedOnFailure() throws
Exception {
+ // removeOnFailure=false keeps the message id of a failed exchange, so
the redelivery is a duplicate
+ getMockEndpoint("mock:result").expectedBodiesReceived("one");
+
+ send("direct:keepOnFailure", true);
+ assertTrue(repository.contains("1"));
+ send("direct:keepOnFailure", false);
+
+ assertMockEndpointsSatisfied();
+ }
+
+ private void assertRedeliveryProcessed(String uri) throws Exception {
+ getMockEndpoint("mock:result").expectedBodiesReceived("one", "one");
+
+ Exchange rolledBack = send(uri, true);
+ assertTrue(rolledBack.isRollbackOnly() ||
rolledBack.isRollbackOnlyLast());
+ assertNull(rolledBack.getException());
+ assertFalse(repository.contains("1"), "The message id of a rollback
only exchange should be removed");
+
+ // the message is redelivered, and processed this time
+ send(uri, false);
+
+ assertMockEndpointsSatisfied();
+ assertTrue(repository.contains("1"));
+ }
+
+ private Exchange send(String uri, boolean rollback) {
+ return template.send(uri, exchange -> {
+ exchange.getIn().setBody("one");
+ exchange.getIn().setHeader("messageId", "1");
+ exchange.getIn().setHeader("rollback", rollback);
+ });
+ }
+}
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
index 6e3565a50d9d..10ab4517e865 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_24.adoc
@@ -712,6 +712,19 @@ A route that relies on a regular expression as token, such
as `\s+` or `(\W+)\s*
In the XML and YAML DSL, a token with the escapes `\n`, `\r` or `\t` (for
example `\r\n`) is turned into the new line,
carriage return and tab characters, so it works as before. Before, only a
leading `\n` was turned into a new line.
+=== camel-core - Idempotent Consumer: an exchange marked rollback only removes
its message id
+
+The xref:components:eips:idempotentConsumer-eip.adoc[Idempotent Consumer] now
treats an exchange marked with
+`markRollbackOnly()` or `markRollbackOnlyLast()` (without an exception) as
failed, as the SQL consumer already does:
+with `removeOnFailure=true` (the default) its message id is removed from the
idempotent repository, so a redelivered
+message is processed instead of being dropped as a duplicate. Before, the
message id was kept.
+
+With a consumer that does not redeliver a rollback only exchange, such as the
file consumer, a later message with the
+same id is now processed instead of being dropped as a duplicate.
+
+To keep the message id, set `removeOnFailure=false`. Note that this also keeps
the message id of an exchange that
+failed with an exception.
+
=== camel-cyberark-vault - credentials are checked at endpoint start, and a
rejected token is retried only once
A `cyberark-vault` endpoint now fails to start when it has no usable
credentials, with a message that names the
