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 f80b68052fceb12a61900076aac8189334005c52 Author: smjain <[email protected]> AuthorDate: Thu Sep 24 08:03:57 2026 +0530 CAMEL-24953: camel-core - Use KeyValueIdempotentRepository in the new test and always release the blocked exchange MemoryIdempotentRepository is deprecated since 4.23. Also rewraps a long comment. Co-Authored-By: Claude Opus 5.5 <[email protected]> --- .../processor/idempotent/IdempotentConsumer.java | 3 ++- .../IdempotentConsumerFailedDuplicateTest.java | 30 ++++++++++++---------- 2 files changed, 18 insertions(+), 15 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 28c7f479152d..1cc89c6c71a3 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 @@ -176,7 +176,8 @@ public class IdempotentConsumer extends BaseProcessorSupport } 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 + // 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); } } 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 index 1fa0885aedc4..9041f86afd21 100644 --- 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 @@ -25,7 +25,7 @@ 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.apache.camel.support.KeyValueIdempotentRepository; import org.junit.jupiter.api.Test; import static org.junit.jupiter.api.Assertions.assertEquals; @@ -39,7 +39,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue; */ class IdempotentConsumerFailedDuplicateTest extends ContextTestSupport { - private final IdempotentRepository repo = MemoryIdempotentRepository.memoryIdempotentRepository(200); + private final IdempotentRepository repo = new KeyValueIdempotentRepository(); private final CountDownLatch firstInProgress = new CountDownLatch(1); private final CountDownLatch releaseFirst = new CountDownLatch(1); @@ -81,18 +81,20 @@ class IdempotentConsumerFailedDuplicateTest extends ContextTestSupport { 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(); + try { + 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)); + } finally { + releaseFirst.countDown(); + } Exchange out = first.get(10, TimeUnit.SECONDS); assertFalse(out.isFailed()); assertNull(out.getProperty(Exchange.DUPLICATE_MESSAGE));
