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 ab21f790a87d CAMEL-25017: camel-core - Claim Check EIP: exchange
copies must not share the claim check repository (#26882)
ab21f790a87d is described below
commit ab21f790a87d3483f6be23dd4b8150a8dc62e3e4
Author: allthingssecurity <[email protected]>
AuthorDate: Mon Sep 28 13:39:35 2026 +0530
CAMEL-25017: camel-core - Claim Check EIP: exchange copies must not share
the claim check repository (#26882)
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../engine/PooledProcessorExchangeFactory.java | 12 ++
.../camel/processor/ClaimCheckProcessor.java | 22 +++-
.../processor/DefaultClaimCheckRepository.java | 26 +++-
.../processor/ClaimCheckEipSplitParallelTest.java | 146 +++++++++++++++++++++
.../processor/ClaimCheckEipStoreCopyTest.java | 85 ++++++++++++
...oledExchangeClaimCheckEipSplitParallelTest.java | 38 ++++++
.../org/apache/camel/util/ExchangeHelperTest.java | 22 ++++
.../org/apache/camel/support/AbstractExchange.java | 3 +
.../org/apache/camel/support/ExchangeHelper.java | 10 ++
.../ROOT/pages/camel-4x-upgrade-guide-4_23.adoc | 14 ++
10 files changed, 370 insertions(+), 8 deletions(-)
diff --git
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/PooledProcessorExchangeFactory.java
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/PooledProcessorExchangeFactory.java
index 17c1511faf8f..2a38eecff745 100644
---
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/PooledProcessorExchangeFactory.java
+++
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/PooledProcessorExchangeFactory.java
@@ -22,6 +22,7 @@ import org.apache.camel.ExchangePattern;
import org.apache.camel.ExchangePropertyKey;
import org.apache.camel.PooledExchange;
import org.apache.camel.Processor;
+import org.apache.camel.SafeCopyProperty;
import org.apache.camel.spi.ProcessorExchangeFactory;
import org.apache.camel.support.DefaultPooledExchange;
import org.apache.camel.support.ExchangeHelper;
@@ -78,6 +79,7 @@ public class PooledProcessorExchangeFactory extends
PrototypeProcessorExchangeFa
// reset the copy's clock for reuse
((ResetableClock) answer.getClock()).reset();
ExchangeHelper.copyResults(answer, exchange);
+ safeCopyClaimCheckRepository(answer);
return answer;
}
@@ -103,6 +105,7 @@ public class PooledProcessorExchangeFactory extends
PrototypeProcessorExchangeFa
((ResetableClock) answer.getClock()).reset();
ExchangeHelper.copyResults(answer, exchange);
+ safeCopyClaimCheckRepository(answer);
// do not reuse message id on copy
answer.getIn().setMessageId(null);
if (handover) {
@@ -114,6 +117,15 @@ public class PooledProcessorExchangeFactory extends
PrototypeProcessorExchangeFa
return answer;
}
+ private static void safeCopyClaimCheckRepository(Exchange copy) {
+ // the claim check repository is scoped per exchange, so the copy must
not share it
+ // (the same as Exchange.copy() does)
+ Object repo =
copy.getProperty(ExchangePropertyKey.CLAIM_CHECK_REPOSITORY);
+ if (repo instanceof SafeCopyProperty scp) {
+ copy.setProperty(ExchangePropertyKey.CLAIM_CHECK_REPOSITORY,
scp.safeCopy());
+ }
+ }
+
@Override
public Exchange create(Endpoint fromEndpoint, ExchangePattern
exchangePattern) {
Exchange answer = pool.poll();
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/ClaimCheckProcessor.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/ClaimCheckProcessor.java
index a0b38b27925b..25fa9a7880b6 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/ClaimCheckProcessor.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/ClaimCheckProcessor.java
@@ -167,8 +167,7 @@ public class ClaimCheckProcessor extends
BaseProcessorSupport implements IdAware
}
private static void operationPushHandler(Exchange exchange, String
claimKey, ClaimCheckRepository repo) {
- // copy exchange, and do not share the unit of work
- Exchange copy = ExchangeHelper.createCorrelatedCopy(exchange, false);
+ Exchange copy = createClaimCheckCopy(exchange);
LOG.debug("Push: {} -> {}", claimKey, copy);
repo.push(copy);
}
@@ -198,9 +197,7 @@ public class ClaimCheckProcessor extends
BaseProcessorSupport implements IdAware
}
private static void operationSetHandler(Exchange exchange, String
claimKey, ClaimCheckRepository repo) {
-
- // copy exchange, and do not share the unit of work
- Exchange copy = ExchangeHelper.createCorrelatedCopy(exchange, false);
+ Exchange copy = createClaimCheckCopy(exchange);
boolean addedNew = repo.add(claimKey, copy);
if (addedNew) {
LOG.debug("Add: {} -> {}", claimKey, copy);
@@ -209,6 +206,21 @@ public class ClaimCheckProcessor extends
BaseProcessorSupport implements IdAware
}
}
+ private static Exchange createClaimCheckCopy(Exchange exchange) {
+ // the stored copy must not carry the repository: it does not need it,
and it must not replace the repository of
+ // the exchange when an aggregation strategy returns the stored copy
as the result. So detach the repository
+ // while copying, instead of copying it (which would copy all its
claim checks on every Set or Push)
+ Object repo =
exchange.removeProperty(ExchangePropertyKey.CLAIM_CHECK_REPOSITORY);
+ try {
+ // copy exchange, and do not share the unit of work
+ return ExchangeHelper.createCorrelatedCopy(exchange, false);
+ } finally {
+ if (repo != null) {
+
exchange.setProperty(ExchangePropertyKey.CLAIM_CHECK_REPOSITORY, repo);
+ }
+ }
+ }
+
private static ClaimCheckRepository getClaimCheckRepository(Exchange
exchange) {
ClaimCheckRepository repo
=
exchange.getProperty(ExchangePropertyKey.CLAIM_CHECK_REPOSITORY,
ClaimCheckRepository.class);
diff --git
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/DefaultClaimCheckRepository.java
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/DefaultClaimCheckRepository.java
index d86b8bf18991..2056df18e1c2 100644
---
a/core/camel-core-processor/src/main/java/org/apache/camel/processor/DefaultClaimCheckRepository.java
+++
b/core/camel-core-processor/src/main/java/org/apache/camel/processor/DefaultClaimCheckRepository.java
@@ -22,15 +22,35 @@ import java.util.HashMap;
import java.util.Map;
import org.apache.camel.Exchange;
+import org.apache.camel.SafeCopyProperty;
import org.apache.camel.spi.ClaimCheckRepository;
/**
* The default {@link ClaimCheckRepository} implementation that is an
in-memory storage.
+ * <p/>
+ * The repository is not thread-safe, as it is scoped per exchange. When an
exchange is copied (such as by the Splitter,
+ * Multicast, Recipient List or Wire Tap EIPs), then the copy gets its own
repository, which starts with the same claim
+ * checks, via {@link #safeCopy()}.
*/
-public class DefaultClaimCheckRepository implements ClaimCheckRepository {
+public class DefaultClaimCheckRepository implements ClaimCheckRepository,
SafeCopyProperty {
- private final Map<String, Exchange> map = new HashMap<>();
- private final Deque<Exchange> stack = new ArrayDeque<>();
+ private final Map<String, Exchange> map;
+ private final Deque<Exchange> stack;
+
+ public DefaultClaimCheckRepository() {
+ this.map = new HashMap<>();
+ this.stack = new ArrayDeque<>();
+ }
+
+ private DefaultClaimCheckRepository(DefaultClaimCheckRepository source) {
+ this.map = new HashMap<>(source.map);
+ this.stack = new ArrayDeque<>(source.stack);
+ }
+
+ @Override
+ public DefaultClaimCheckRepository safeCopy() {
+ return new DefaultClaimCheckRepository(this);
+ }
@Override
public boolean add(String key, Exchange exchange) {
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/ClaimCheckEipSplitParallelTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/ClaimCheckEipSplitParallelTest.java
new file mode 100644
index 000000000000..ba9d9c1e0b1b
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/ClaimCheckEipSplitParallelTest.java
@@ -0,0 +1,146 @@
+/*
+ * 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.TimeUnit;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.model.ClaimCheckOperation;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The claim check repository is scoped per exchange: the exchanges created by
a parallel Splitter get their own copy of
+ * the parent's claim checks, and must not get the message of another part
back.
+ */
+class ClaimCheckEipSplitParallelTest extends ContextTestSupport {
+
+ private final CountDownLatch part0Saved = new CountDownLatch(1);
+ private final CountDownLatch part1Saved = new CountDownLatch(1);
+ private final CountDownLatch part0Restored = new CountDownLatch(1);
+
+ @Test
+ void testSetGetInParallelSplit() throws Exception {
+ assertEachPartRestoresItsOwnMessage("direct:setget");
+ }
+
+ @Test
+ void testPushPopInParallelSplit() throws Exception {
+ assertEachPartRestoresItsOwnMessage("direct:pushpop");
+ }
+
+ @Test
+ void testPartGetsClaimCheckOfParent() throws Exception {
+ getMockEndpoint("mock:part").expectedBodiesReceived("A,B", "A,B");
+ getMockEndpoint("mock:result").expectedBodiesReceived("A,B");
+
+ template.sendBody("direct:parent", "A,B");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ private void assertEachPartRestoresItsOwnMessage(String uri) throws
Exception {
+ getMockEndpoint("mock:part").expectedBodiesReceivedInAnyOrder("0:A",
"1:B");
+ getMockEndpoint("mock:result").expectedBodiesReceived("A,B");
+
+ template.sendBody(uri, "A,B");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ // the parts save their message in this order: part 0, part 1, and then
restore it: part 0, part 1
+
+ private void beforeSave(Exchange exchange) throws Exception {
+ if (splitIndex(exchange) == 1) {
+ await(part0Saved);
+ }
+ }
+
+ private void afterSave(Exchange exchange) throws Exception {
+ if (splitIndex(exchange) == 0) {
+ part0Saved.countDown();
+ await(part1Saved);
+ } else {
+ part1Saved.countDown();
+ await(part0Restored);
+ }
+ }
+
+ private void afterRestore(Exchange exchange) {
+ if (splitIndex(exchange) == 0) {
+ part0Restored.countDown();
+ }
+ }
+
+ private static int splitIndex(Exchange exchange) {
+ return exchange.getProperty(Exchange.SPLIT_INDEX, Integer.class);
+ }
+
+ private static void await(CountDownLatch latch) throws
InterruptedException {
+ assertTrue(latch.await(10, TimeUnit.SECONDS), "The other part did not
reach the expected step");
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:setget")
+ .claimCheck(ClaimCheckOperation.Set, "original")
+ .split(body().tokenize(",")).parallelProcessing()
+ .process(e -> beforeSave(e))
+ .claimCheck(ClaimCheckOperation.Set, "item")
+ .process(e -> afterSave(e))
+ .setBody(constant("reply-from-service"))
+ .claimCheck(ClaimCheckOperation.Get, "item")
+ .process(e -> afterRestore(e))
+
.setBody(simple("${exchangeProperty.CamelSplitIndex}:${body}"))
+ .to("mock:part")
+ .end()
+ .claimCheck(ClaimCheckOperation.Get, "original")
+ .to("mock:result");
+
+ from("direct:pushpop")
+ .claimCheck(ClaimCheckOperation.Set, "original")
+ .split(body().tokenize(",")).parallelProcessing()
+ .process(e -> beforeSave(e))
+ .claimCheck(ClaimCheckOperation.Push)
+ .process(e -> afterSave(e))
+ .setBody(constant("reply-from-service"))
+ .claimCheck(ClaimCheckOperation.Pop)
+ .process(e -> afterRestore(e))
+
.setBody(simple("${exchangeProperty.CamelSplitIndex}:${body}"))
+ .to("mock:part")
+ .end()
+ .claimCheck(ClaimCheckOperation.Get, "original")
+ .to("mock:result");
+
+ from("direct:parent")
+ .claimCheck(ClaimCheckOperation.Set, "original")
+ .split(body().tokenize(","))
+ .claimCheck(ClaimCheckOperation.Get, "original")
+ .to("mock:part")
+ .end()
+ .to("mock:result");
+ }
+ };
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/ClaimCheckEipStoreCopyTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/ClaimCheckEipStoreCopyTest.java
new file mode 100644
index 000000000000..e74fc4e31c43
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/ClaimCheckEipStoreCopyTest.java
@@ -0,0 +1,85 @@
+/*
+ * 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.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.ExchangePropertyKey;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.model.ClaimCheckOperation;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The copy of the exchange that Set and Push store does not carry the claim
check repository, and making it must not
+ * copy the repository (which would copy all its claim checks on every Set or
Push).
+ */
+class ClaimCheckEipStoreCopyTest extends ContextTestSupport {
+
+ private final CountingClaimCheckRepository repository = new
CountingClaimCheckRepository();
+ private final AtomicInteger safeCopiesBeforeMock = new AtomicInteger(-1);
+ private final AtomicReference<Object> repositoryAtEnd = new
AtomicReference<>();
+
+ @Test
+ void testSetAndPushDoNotCopyTheRepository() throws Exception {
+ getMockEndpoint("mock:result").expectedBodiesReceived("Hello World");
+
+ template.sendBody("direct:start", "Hello World");
+
+ assertMockEndpointsSatisfied();
+ assertEquals(0, safeCopiesBeforeMock.get(), "Set and Push should not
copy the claim check repository");
+ assertSame(repository, repositoryAtEnd.get(), "The exchange should
keep its claim check repository");
+ assertTrue(repository.contains("a") && repository.contains("b"), "The
repository should keep the claim checks");
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start")
+ .process(e ->
e.setProperty(ExchangePropertyKey.CLAIM_CHECK_REPOSITORY, repository))
+ .claimCheck(ClaimCheckOperation.Set, "a")
+ .claimCheck(ClaimCheckOperation.Set, "b")
+ .claimCheck(ClaimCheckOperation.Push)
+ .claimCheck(ClaimCheckOperation.Pop)
+ .process(e -> {
+ // capture before the mock endpoint, which copies
the exchange
+
safeCopiesBeforeMock.set(repository.safeCopies.get());
+
repositoryAtEnd.set(e.getProperty(ExchangePropertyKey.CLAIM_CHECK_REPOSITORY));
+ })
+ .to("mock:result");
+ }
+ };
+ }
+
+ private static final class CountingClaimCheckRepository extends
DefaultClaimCheckRepository {
+
+ private final AtomicInteger safeCopies = new AtomicInteger();
+
+ @Override
+ public DefaultClaimCheckRepository safeCopy() {
+ safeCopies.incrementAndGet();
+ return super.safeCopy();
+ }
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/PooledExchangeClaimCheckEipSplitParallelTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/PooledExchangeClaimCheckEipSplitParallelTest.java
new file mode 100644
index 000000000000..78585a18372d
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/PooledExchangeClaimCheckEipSplitParallelTest.java
@@ -0,0 +1,38 @@
+/*
+ * 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.CamelContext;
+import org.apache.camel.ExtendedCamelContext;
+import org.apache.camel.impl.engine.PooledExchangeFactory;
+import org.apache.camel.impl.engine.PooledProcessorExchangeFactory;
+
+/**
+ * The same as {@link ClaimCheckEipSplitParallelTest} with pooled exchanges,
where the sub-exchanges are created by
+ * {@link PooledProcessorExchangeFactory}.
+ */
+class PooledExchangeClaimCheckEipSplitParallelTest extends
ClaimCheckEipSplitParallelTest {
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext camelContext = super.createCamelContext();
+ ExtendedCamelContext ecc = camelContext.getCamelContextExtension();
+ ecc.setExchangeFactory(new PooledExchangeFactory());
+ ecc.setProcessorExchangeFactory(new PooledProcessorExchangeFactory());
+ return camelContext;
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/util/ExchangeHelperTest.java
b/core/camel-core/src/test/java/org/apache/camel/util/ExchangeHelperTest.java
index c2c03180efd1..14de107bf3b9 100644
---
a/core/camel-core/src/test/java/org/apache/camel/util/ExchangeHelperTest.java
+++
b/core/camel-core/src/test/java/org/apache/camel/util/ExchangeHelperTest.java
@@ -23,12 +23,15 @@ import java.util.Map;
import org.apache.camel.ContextTestSupport;
import org.apache.camel.Exchange;
import org.apache.camel.ExchangePattern;
+import org.apache.camel.ExchangePropertyKey;
import org.apache.camel.Message;
import org.apache.camel.NoSuchBeanException;
import org.apache.camel.NoSuchHeaderException;
import org.apache.camel.NoSuchPropertyException;
import org.apache.camel.converter.stream.InputStreamCache;
import org.apache.camel.impl.DefaultCamelContext;
+import org.apache.camel.processor.DefaultClaimCheckRepository;
+import org.apache.camel.spi.ClaimCheckRepository;
import org.apache.camel.support.DefaultExchange;
import org.apache.camel.support.DefaultMessage;
import org.apache.camel.support.ExchangeHelper;
@@ -299,6 +302,25 @@ public class ExchangeHelperTest extends ContextTestSupport
{
assertNull(in.getExchange());
}
+ @Test
+ public void
testCopyExchangeWithPropertiesDoesNotShareClaimCheckRepository() {
+ DefaultClaimCheckRepository repo = new DefaultClaimCheckRepository();
+ Exchange stored = new DefaultExchange(context);
+ repo.add("original", stored);
+ exchange.setProperty(ExchangePropertyKey.CLAIM_CHECK_REPOSITORY, repo);
+
+ Exchange copy = ExchangeHelper.copyExchangeWithProperties(exchange,
context);
+
+ ClaimCheckRepository copyRepo
+ = copy.getProperty(ExchangePropertyKey.CLAIM_CHECK_REPOSITORY,
ClaimCheckRepository.class);
+ assertNotSame(repo, copyRepo, "The copy should get its own claim check
repository");
+ assertSame(stored, copyRepo.get("original"), "The copy should get the
claim checks of the exchange");
+
+ copyRepo.add("part", new DefaultExchange(context));
+ assertFalse(repo.contains("part"), "A claim check stored by the copy
should not be visible to the exchange");
+ assertSame(repo,
exchange.getProperty(ExchangePropertyKey.CLAIM_CHECK_REPOSITORY));
+ }
+
@Override
@BeforeEach
public void setUp() throws Exception {
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/AbstractExchange.java
b/core/camel-support/src/main/java/org/apache/camel/support/AbstractExchange.java
index fdaa2c467fe6..102bee085166 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/AbstractExchange.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/AbstractExchange.java
@@ -175,6 +175,9 @@ abstract class AbstractExchange implements Exchange,
ExchangeExtension {
exchange.internalProperties.computeIfPresent(ExchangePropertyKey.MESSAGE_HISTORY,
(k, v) -> new
CopyOnWriteArrayList<>((List<MessageHistory>) v));
}
+ // the claim check repository is scoped per exchange, so the copy must
not share it
+
exchange.internalProperties.computeIfPresent(ExchangePropertyKey.CLAIM_CHECK_REPOSITORY,
+ (k, v) -> v instanceof SafeCopyProperty scp ? scp.safeCopy() :
v);
return exchange;
}
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/ExchangeHelper.java
b/core/camel-support/src/main/java/org/apache/camel/support/ExchangeHelper.java
index d7ecd3df23f8..773a7335e4f6 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/ExchangeHelper.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/ExchangeHelper.java
@@ -49,6 +49,7 @@ import org.apache.camel.NoSuchPropertyException;
import org.apache.camel.NoTypeConversionAvailableException;
import org.apache.camel.Route;
import org.apache.camel.RuntimeCamelException;
+import org.apache.camel.SafeCopyProperty;
import org.apache.camel.StreamCache;
import org.apache.camel.TypeConversionException;
import org.apache.camel.VariableAware;
@@ -844,6 +845,14 @@ public final class ExchangeHelper {
}
}
+ private static void setClaimCheckRepository(Exchange target, Exchange
source) {
+ // the claim check repository is scoped per exchange, so the copy must
not share it (as in Exchange.copy())
+ final Object repo =
source.getProperty(ExchangePropertyKey.CLAIM_CHECK_REPOSITORY);
+ if (repo instanceof SafeCopyProperty scp) {
+ target.setProperty(ExchangePropertyKey.CLAIM_CHECK_REPOSITORY,
scp.safeCopy());
+ }
+ }
+
/**
* Copies the exchange but the copy will be tied to the given context
*
@@ -854,6 +863,7 @@ public final class ExchangeHelper {
Exchange answer =
exchange.getExchangeExtension().createCopyWithProperties(context);
setMessageHistory(answer, exchange);
+ setClaimCheckRepository(answer, exchange);
answer.setIn(exchange.getIn().copy());
if (exchange.hasOut()) {
diff --git
a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
index b3546fb8648a..e16f9a2740bc 100644
--- a/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
+++ b/docs/user-manual/modules/ROOT/pages/camel-4x-upgrade-guide-4_23.adoc
@@ -847,6 +847,20 @@ unknown property of a bean created through its builder
names the properties the
`PropertyBindingSupport.setPropertiesOnTarget` now reports an unknown property
with the property name and value, as
it rethrew the `PropertyBindingException` with its null cause since 4.21,
which surfaced as `NullPointerException: e`.
+=== camel-core - Claim Check EIP - exchange copies get their own claim check
repository
+
+The Claim Check EIP keeps its claim checks in a repository that belongs to the
exchange. The javadoc of
+`ClaimCheckProcessor` already says that the repository is "not shared among
Exchanges", but a copy of an exchange
+used the same repository instance as the exchange it was copied from. This
only happened when the exchange had used
+the Claim Check before the copy was made, because the repository is created on
first use. The exchanges created by
+the Splitter, Multicast, Recipient List and Wire Tap EIPs, and the exchanges
sent to a SEDA or Disruptor queue, then shared the
+repository of the parent, so with parallel processing a Get or Pop could
return the message of another part.
+
+Such a copy now gets its own copy of the repository. It starts with the claim
checks of the parent, so a part can
+still get what the parent stored, but what a part stores, removes, pushes or
pops is no longer visible to the parent
+or to the other parts. After a Multicast or Recipient List, the parent
continues with the repository of the result
+exchange (the exchange that the aggregation strategy returned), as it already
did for the other exchange properties.
+
=== camel-core - Splitter unwraps Exchange parts
The Splitter EIP now unwraps parts of type `org.apache.camel.Exchange`, the
same way it already unwrapped parts of
