This is an automated email from the ASF dual-hosted git repository. apupier pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/camel.git
commit 64e17e003b0ca0d559dd9b689e5185256b5e7c61 Author: smjain <[email protected]> AuthorDate: Wed Sep 23 19:11:03 2026 +0530 CAMEL-24953: camel-core - Idempotent Consumer: do not let a routed duplicate remove or confirm the key With skipDuplicate=false a duplicate is routed on with the CamelDuplicateMessage property set. It was given the same IdempotentOnCompletion as a new message, so when the duplicate failed (with the default removeOnFailure=true) the completion removed the key from the repository. That key was added by the original exchange, which had already completed successfully or was still in flight. The next copy of the message was then processed again, and in eager mode it could even be processed concurrently with the original exchange. On success the duplicate also confirmed (and in non-eager mode added) a key it did not own. A duplicate never added the key, so there is nothing for it to roll back or confirm. Only register the idempotent completion for new messages; a duplicate that is not skipped is routed on without it. New messages are not affected. Co-Authored-By: Claude Opus 5.5 <[email protected]> --- .../processor/idempotent/IdempotentConsumer.java | 26 ++-- .../IdempotentConsumerFailedDuplicateTest.java | 157 +++++++++++++++++++++ 2 files changed, 172 insertions(+), 11 deletions(-) diff --git a/core/camel-core-processor/src/main/java/org/apache/camel/processor/idempotent/IdempotentConsumer.java b/core/camel-core-processor/src/main/java/org/apache/camel/processor/idempotent/IdempotentConsumer.java index 71e75adbb51e..28c7f479152d 100644 --- a/core/camel-core-processor/src/main/java/org/apache/camel/processor/idempotent/IdempotentConsumer.java +++ b/core/camel-core-processor/src/main/java/org/apache/camel/processor/idempotent/IdempotentConsumer.java @@ -162,19 +162,23 @@ public class IdempotentConsumer extends BaseProcessorSupport callback.done(true); return true; } - } - - final Synchronization onCompletion - = new IdempotentOnCompletion(idempotentRepository, messageId, eager, removeOnFailure); - if (completionEager) { - // the callback will eager complete - target = new IdempotentConsumerCallback(exchange, onCompletion, callback); - } else { - // we can use existing callback as target + // the duplicate is routed on, but it did not add the key (the exchange that did owns it), + // so the duplicate must not confirm the key, nor remove it if it fails target = callback; - // the scope is to do the idempotent completion work as an unit of work on the exchange when its done being routed - exchange.getExchangeExtension().addOnCompletion(onCompletion); + } else { + final Synchronization onCompletion + = new IdempotentOnCompletion(idempotentRepository, messageId, eager, removeOnFailure); + + if (completionEager) { + // the callback will eager complete + target = new IdempotentConsumerCallback(exchange, onCompletion, callback); + } else { + // we can use existing callback as target + target = callback; + // the scope is to do the idempotent completion work as an unit of work on the exchange when its done being routed + exchange.getExchangeExtension().addOnCompletion(onCompletion); + } } } catch (Exception e) { exchange.setException(e); diff --git a/core/camel-core/src/test/java/org/apache/camel/processor/IdempotentConsumerFailedDuplicateTest.java b/core/camel-core/src/test/java/org/apache/camel/processor/IdempotentConsumerFailedDuplicateTest.java new file mode 100644 index 000000000000..f1c71b295903 --- /dev/null +++ b/core/camel-core/src/test/java/org/apache/camel/processor/IdempotentConsumerFailedDuplicateTest.java @@ -0,0 +1,157 @@ +/* + * 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 java.util.concurrent.CountDownLatch; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +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.IdempotentRepository; +import org.apache.camel.support.processor.idempotent.MemoryIdempotentRepository; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * With skipDuplicate=false a duplicate is routed on. If the duplicate fails, it must not remove the key that the + * original exchange added. + */ +public class IdempotentConsumerFailedDuplicateTest extends ContextTestSupport { + + private final IdempotentRepository repo = MemoryIdempotentRepository.memoryIdempotentRepository(200); + private final CountDownLatch firstInProgress = new CountDownLatch(1); + private final CountDownLatch releaseFirst = new CountDownLatch(1); + + @Override + public boolean isUseRouteBuilder() { + return false; + } + + @Test + public void testFailedDuplicateDoesNotRemoveKey() throws Exception { + addRoute(true, false); + assertFailedDuplicateDoesNotRemoveKey(); + } + + @Test + public void testFailedDuplicateDoesNotRemoveKeyCompletionEager() throws Exception { + addRoute(true, true); + assertFailedDuplicateDoesNotRemoveKey(); + } + + @Test + public void testFailedDuplicateDoesNotRemoveKeyNonEager() throws Exception { + addRoute(false, false); + assertFailedDuplicateDoesNotRemoveKey(); + } + + @Test + public void testFailedDuplicateDoesNotRemoveKeyOfInflightExchange() throws Exception { + addRoute(true, false); + + MockEndpoint newMessages = getMockEndpoint("mock:new"); + newMessages.expectedBodiesReceived("first"); + MockEndpoint duplicates = getMockEndpoint("mock:duplicate"); + duplicates.expectedBodiesReceived("second", "third"); + + // the first exchange adds the key (eager) and waits inside the route + Future<Exchange> first = template.asyncSend("direct:start", e -> { + e.getIn().setHeader("messageId", "1"); + e.getIn().setHeader("block", true); + e.getIn().setBody("first"); + }); + assertTrue(firstInProgress.await(10, TimeUnit.SECONDS)); + + // a duplicate that fails while the first exchange is still in progress + Exchange second = send("second"); + assertTrue(second.isFailed()); + assertTrue(repo.contains("1"), "The failed duplicate must not remove the key of the in-flight exchange"); + + // so another copy is still a duplicate, and is not processed concurrently with the first exchange + Exchange third = send("third"); + assertEquals(Boolean.TRUE, third.getProperty(Exchange.DUPLICATE_MESSAGE)); + + releaseFirst.countDown(); + Exchange out = first.get(10, TimeUnit.SECONDS); + assertFalse(out.isFailed()); + assertNull(out.getProperty(Exchange.DUPLICATE_MESSAGE)); + + assertMockEndpointsSatisfied(); + assertTrue(repo.contains("1")); + } + + private void assertFailedDuplicateDoesNotRemoveKey() throws Exception { + MockEndpoint newMessages = getMockEndpoint("mock:new"); + newMessages.expectedBodiesReceived("first"); + MockEndpoint duplicates = getMockEndpoint("mock:duplicate"); + duplicates.expectedBodiesReceived("second", "third"); + + Exchange first = send("first"); + assertFalse(first.isFailed()); + assertTrue(repo.contains("1")); + + Exchange second = send("second"); + assertTrue(second.isFailed()); + assertEquals(Boolean.TRUE, second.getProperty(Exchange.DUPLICATE_MESSAGE)); + assertTrue(repo.contains("1"), "The failed duplicate must not remove the key added by the first exchange"); + + Exchange third = send("third"); + assertEquals(Boolean.TRUE, third.getProperty(Exchange.DUPLICATE_MESSAGE)); + + assertMockEndpointsSatisfied(); + } + + private Exchange send(String body) { + return template.send("direct:start", e -> { + e.getIn().setHeader("messageId", "1"); + e.getIn().setBody(body); + }); + } + + private void addRoute(boolean eager, boolean completionEager) throws Exception { + context.addRoutes(new RouteBuilder() { + @Override + public void configure() { + from("direct:start") + .idempotentConsumer(header("messageId")).idempotentRepository(repo) + .eager(eager).completionEager(completionEager).skipDuplicate(false) + .choice() + .when(exchangeProperty(Exchange.DUPLICATE_MESSAGE).isEqualTo(true)) + .to("mock:duplicate") + .throwException(new IllegalStateException("Cannot handle the duplicate")) + .otherwise() + .process(e -> { + if (e.getIn().getHeader("block", false, Boolean.class)) { + firstInProgress.countDown(); + releaseFirst.await(10, TimeUnit.SECONDS); + } + }) + .to("mock:new") + .end() + .end(); + } + }); + context.start(); + } +}
