This is an automated email from the ASF dual-hosted git repository.
Croway 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 9eaa29827812 CAMEL-24800: camel-core - pooled exchange copies must own
their clock (#26553)
9eaa29827812 is described below
commit 9eaa29827812b851335b60f3c7d7c30de4da95d4
Author: Federico Mariani <[email protected]>
AuthorDate: Thu Sep 17 19:13:51 2026 +0200
CAMEL-24800: camel-core - pooled exchange copies must own their clock
(#26553)
* CAMEL-24800: camel-core - pooled exchange copies must own their clock
DefaultPooledExchange(Exchange parent) reused the parent's ResetableClock
when the parent already had one. Splitter children are pooled copies of
the Splitter's working copy, so parent and child shared one clock: when
the working copy was released first, its done() unset the created time,
and the child's done() then skipped the reset entirely. The dirty child
went back into the pool, leaking exchange properties and variables into
the next message's split children.
Every pooled copy now gets its own ResetableClock.
Co-Authored-By: Claude Fable 5.1 <[email protected]>
* CAMEL-24800: assert the first child carries the state before checking it
does not leak
Co-Authored-By: Claude Fable 5.1 <[email protected]>
---------
Co-authored-by: Claude Fable 5.1 <[email protected]>
---
.../processor/PooledExchangeSplitLeakTest.java | 78 ++++++++++++++++++++++
.../camel/support/DefaultPooledExchange.java | 10 +--
2 files changed, 81 insertions(+), 7 deletions(-)
diff --git
a/core/camel-core/src/test/java/org/apache/camel/processor/PooledExchangeSplitLeakTest.java
b/core/camel-core/src/test/java/org/apache/camel/processor/PooledExchangeSplitLeakTest.java
new file mode 100644
index 000000000000..2e08a7d72ca1
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/processor/PooledExchangeSplitLeakTest.java
@@ -0,0 +1,78 @@
+/*
+ * 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.List;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.ExtendedCamelContext;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.engine.PooledExchangeFactory;
+import org.apache.camel.impl.engine.PooledProcessorExchangeFactory;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+public class PooledExchangeSplitLeakTest extends ContextTestSupport {
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext camelContext = super.createCamelContext();
+ ExtendedCamelContext ecc = camelContext.getCamelContextExtension();
+ ecc.setExchangeFactory(new PooledExchangeFactory());
+ ecc.setProcessorExchangeFactory(new PooledProcessorExchangeFactory());
+ return camelContext;
+ }
+
+ @Test
+ public void testSplitChildrenDoNotLeakStateBetweenMessages() throws
Exception {
+ MockEndpoint mock = getMockEndpoint("mock:split");
+ mock.expectedMessageCount(2);
+
+ template.sendBody("direct:start", List.of("first"));
+ template.sendBody("direct:start", List.of("second"));
+
+ mock.assertIsSatisfied();
+
+ // the child of the first message carries the state set by the
processor
+ assertEquals("from-first",
mock.getExchanges().get(0).getProperty("leakProp"));
+ assertEquals("from-first",
mock.getExchanges().get(0).getVariable("leak"));
+ // the child of the second message is a reused pooled exchange; it
must not carry the first child's state
+ assertNull(mock.getExchanges().get(1).getProperty("leakProp"));
+ assertNull(mock.getExchanges().get(1).getVariable("leak"));
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ public void configure() {
+ from("direct:start").split(body())
+ .process(e -> {
+ if ("first".equals(e.getMessage().getBody())) {
+ e.setProperty("leakProp", "from-first");
+ e.setVariable("leak", "from-first");
+ }
+ })
+ .to("mock:split");
+ }
+ };
+ }
+}
diff --git
a/core/camel-support/src/main/java/org/apache/camel/support/DefaultPooledExchange.java
b/core/camel-support/src/main/java/org/apache/camel/support/DefaultPooledExchange.java
index 9397fe32455f..89fb75996920 100644
---
a/core/camel-support/src/main/java/org/apache/camel/support/DefaultPooledExchange.java
+++
b/core/camel-support/src/main/java/org/apache/camel/support/DefaultPooledExchange.java
@@ -45,13 +45,9 @@ public final class DefaultPooledExchange extends
AbstractExchange implements Poo
super(parent);
this.originalPattern = parent.getPattern();
- Clock parentClock = parent.getClock();
-
- if (parentClock instanceof ResetableClock rs) {
- this.clock = rs;
- } else {
- this.clock = new ResetableClock(parent.getClock());
- }
+ // the copy must own its clock: sharing the parent's clock makes
done() on the parent unset the
+ // created time of the copy as well, and the copy is then returned to
the pool without being reset
+ this.clock = new ResetableClock(parent.getClock());
}
public DefaultPooledExchange(CamelContext context, ExchangePattern
pattern) {
