This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch camel-4.22.x
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/camel-4.22.x by this push:
new 6d6739dc2572 [backport camel-4.22.x] CAMEL-25559: camel-core -
PooledExchangeFactory puts an exchange that was given back without being routed
into the pool twice (#27741)
6d6739dc2572 is described below
commit 6d6739dc2572700d950abf02bcee5ca528727abe
Author: Guillaume Nodet <[email protected]>
AuthorDate: Tue Oct 13 07:58:29 2026 +0200
[backport camel-4.22.x] CAMEL-25559: camel-core - PooledExchangeFactory
puts an exchange that was given back without being routed into the pool twice
(#27741)
Backport of #27698. With the pooled exchange factory, a consumer that gave
back
an exchange it did not route put it into the pool twice, so the next two
messages shared one exchange object. PooledExchangeFactory.release now
offers an
auto-release exchange to the pool exactly once.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---
.../camel/impl/engine/PooledExchangeFactory.java | 55 +++++-
...tEagerMaxMessagesPerPollPooledExchangeTest.java | 69 ++++++++
...nsumeReadLockNotAcquiredPooledExchangeTest.java | 146 ++++++++++++++++
.../PooledExchangeFactoryReleaseNotRoutedTest.java | 188 +++++++++++++++++++++
4 files changed, 451 insertions(+), 7 deletions(-)
diff --git
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/PooledExchangeFactory.java
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/PooledExchangeFactory.java
index 24392c5f280b..c7861ddb1613 100644
---
a/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/PooledExchangeFactory.java
+++
b/core/camel-base-engine/src/main/java/org/apache/camel/impl/engine/PooledExchangeFactory.java
@@ -93,13 +93,49 @@ public final class PooledExchangeFactory extends
PrototypeExchangeFactory {
return exchange;
}
+ /**
+ * Releases the exchange back into the pool.
+ * <p/>
+ * An exchange created with auto release is put into the pool when it is
done. If it is not done yet (it was given
+ * back without being routed) this method does that and returns whether
the pool took it, so false when the pool was
+ * full and the exchange was discarded. If it is already done, its unit of
work has handed it to the pool before,
+ * and this method does nothing and returns true; whether the pool took it
at that time is not known here.
+ */
@Override
public boolean release(Exchange exchange) {
try {
// done exchange before returning to pool
PooledExchange ee = (PooledExchange) exchange;
- ee.done();
+ if (ee.isAutoRelease()) {
+ // an auto-release exchange is offered to the pool once: by
its on-done task when the unit of work is
+ // done, or here when the exchange was given back without
being routed. Offering it in both places
+ // would put it in the pool twice.
+ if (ee.getClock().getCreated() <= 0) {
+ // already done (the clock is unset by done), so the
on-done task has offered it
+ return true;
+ }
+ // done() would run the on-done task and offer the exchange,
but the result of that offer cannot be
+ // read from here. The caller owns the exchange until it is
offered, so take the task off for this
+ // call, offer the exchange here, and put the task back before
the offer makes the exchange visible
+ // to other threads.
+ ee.onDone(null);
+ try {
+ ee.done();
+ } finally {
+ ee.onDone(onDone);
+ }
+ } else {
+ ee.done();
+ }
+ } catch (Exception e) {
+ return discard(exchange);
+ }
+ return doRelease(exchange);
+ }
+
+ private boolean doRelease(Exchange exchange) {
+ try {
// only release back in pool if reset was success
boolean inserted = pool.offer(exchange);
@@ -112,12 +148,16 @@ public final class PooledExchangeFactory extends
PrototypeExchangeFactory {
}
return inserted;
} catch (Exception e) {
- if (statisticsEnabled) {
- statistics.discarded.increment();
- }
- LOG.debug("Error resetting exchange: {}. This exchange is
discarded.", exchange);
- return false;
+ return discard(exchange);
+ }
+ }
+
+ private boolean discard(Exchange exchange) {
+ if (statisticsEnabled) {
+ statistics.discarded.increment();
}
+ LOG.debug("Error resetting exchange: {}. This exchange is discarded.",
exchange);
+ return false;
}
private PooledExchange createPooledExchange(Endpoint fromEndpoint, boolean
autoRelease) {
@@ -157,7 +197,8 @@ public final class PooledExchangeFactory extends
PrototypeExchangeFactory {
@Override
public void onDone(Exchange exchange) {
- release(exchange);
+ // the exchange is done (reset) when this is called
+ doRelease(exchange);
}
}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumeNotEagerMaxMessagesPerPollPooledExchangeTest.java
b/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumeNotEagerMaxMessagesPerPollPooledExchangeTest.java
new file mode 100644
index 000000000000..ae3a4abc31a4
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumeNotEagerMaxMessagesPerPollPooledExchangeTest.java
@@ -0,0 +1,69 @@
+/*
+ * 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.component.file;
+
+import org.apache.camel.CamelContext;
+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.impl.engine.PooledExchangeFactory;
+import org.junit.jupiter.api.Test;
+
+/**
+ * With eagerMaxMessagesPerPoll=false the consumer creates an exchange for
every file of the poll and gives back the
+ * ones beyond maxMessagesPerPoll without routing them. With pooled exchanges,
the files of the next poll must each get
+ * their own exchange.
+ */
+class FileConsumeNotEagerMaxMessagesPerPollPooledExchangeTest extends
ContextTestSupport {
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext camelContext = super.createCamelContext();
+ camelContext.getCamelContextExtension().setExchangeFactory(new
PooledExchangeFactory());
+ return camelContext;
+ }
+
+ @Test
+ void filesBeyondMaxMessagesPerPollAreConsumedByTheNextPoll() throws
Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("AAA", "BBB", "CCC", "DDD");
+ mock.expectedHeaderValuesReceivedInAnyOrder(Exchange.FILE_NAME_ONLY,
"a.txt", "b.txt", "c.txt", "d.txt");
+
+ template.sendBodyAndHeader(fileUri(), "AAA", Exchange.FILE_NAME,
"a.txt");
+ template.sendBodyAndHeader(fileUri(), "BBB", Exchange.FILE_NAME,
"b.txt");
+ template.sendBodyAndHeader(fileUri(), "CCC", Exchange.FILE_NAME,
"c.txt");
+ template.sendBodyAndHeader(fileUri(), "DDD", Exchange.FILE_NAME,
"d.txt");
+
+ context.getRouteController().startRoute("foo");
+
+ assertMockEndpointsSatisfied();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from(fileUri("?initialDelay=0&delay=10&maxMessagesPerPoll=2&eagerMaxMessagesPerPoll=false&sortBy=file:name"))
+ .routeId("foo").autoStartup(false)
+ .convertBodyTo(String.class)
+ .to("mock:result");
+ }
+ };
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumeReadLockNotAcquiredPooledExchangeTest.java
b/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumeReadLockNotAcquiredPooledExchangeTest.java
new file mode 100644
index 000000000000..e42619361377
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/component/file/FileConsumeReadLockNotAcquiredPooledExchangeTest.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.component.file;
+
+import java.io.File;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.LoggingLevel;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.mock.MockEndpoint;
+import org.apache.camel.impl.engine.PooledExchangeFactory;
+import org.apache.camel.spi.Registry;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * The consumer gives back the exchange of every file whose read lock it could
not get, without routing it. With pooled
+ * exchanges, the files of the next poll must each get their own exchange.
+ */
+class FileConsumeReadLockNotAcquiredPooledExchangeTest extends
ContextTestSupport {
+
+ private final RefuseTwiceReadLockStrategy readLockStrategy = new
RefuseTwiceReadLockStrategy();
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext camelContext = super.createCamelContext();
+ camelContext.getCamelContextExtension().setExchangeFactory(new
PooledExchangeFactory());
+ return camelContext;
+ }
+
+ @Override
+ protected Registry createCamelRegistry() throws Exception {
+ Registry registry = super.createCamelRegistry();
+ registry.bind("refuseTwice", readLockStrategy);
+ return registry;
+ }
+
+ @Test
+ void filesWithoutReadLockAreConsumedByTheNextPoll() throws Exception {
+ MockEndpoint mock = getMockEndpoint("mock:result");
+ mock.expectedBodiesReceived("AAA", "BBB");
+ mock.expectedHeaderValuesReceivedInAnyOrder(Exchange.FILE_NAME_ONLY,
"a.txt", "b.txt");
+
+ template.sendBodyAndHeader(fileUri(), "AAA", Exchange.FILE_NAME,
"a.txt");
+ template.sendBodyAndHeader(fileUri(), "BBB", Exchange.FILE_NAME,
"b.txt");
+
+ context.getRouteController().startRoute("foo");
+
+ assertMockEndpointsSatisfied();
+ // both files were refused in the first poll and locked in the second
one
+ assertEquals(4, readLockStrategy.attempts.get());
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+
from(fileUri("?initialDelay=0&delay=10&sortBy=file:name&exclusiveReadLockStrategy=#refuseTwice"))
+ .routeId("foo").autoStartup(false)
+ .convertBodyTo(String.class)
+ .to("mock:result");
+ }
+ };
+ }
+
+ /**
+ * Does not give the lock for the first two files it is asked for (the two
files of the first poll).
+ */
+ private static final class RefuseTwiceReadLockStrategy implements
GenericFileExclusiveReadLockStrategy<File> {
+
+ private final AtomicInteger attempts = new AtomicInteger();
+
+ @Override
+ public void prepareOnStartup(GenericFileOperations<File> operations,
GenericFileEndpoint<File> endpoint) {
+ // noop
+ }
+
+ @Override
+ public boolean acquireExclusiveReadLock(
+ GenericFileOperations<File> operations, GenericFile<File>
file, Exchange exchange) {
+ return attempts.incrementAndGet() > 2;
+ }
+
+ @Override
+ public void releaseExclusiveReadLockOnAbort(
+ GenericFileOperations<File> operations, GenericFile<File>
file, Exchange exchange) {
+ // noop
+ }
+
+ @Override
+ public void releaseExclusiveReadLockOnRollback(
+ GenericFileOperations<File> operations, GenericFile<File>
file, Exchange exchange) {
+ // noop
+ }
+
+ @Override
+ public void releaseExclusiveReadLockOnCommit(
+ GenericFileOperations<File> operations, GenericFile<File>
file, Exchange exchange) {
+ // noop
+ }
+
+ @Override
+ public void setTimeout(long timeout) {
+ // noop
+ }
+
+ @Override
+ public void setCheckInterval(long checkInterval) {
+ // noop
+ }
+
+ @Override
+ public void setReadLockLoggingLevel(LoggingLevel readLockLoggingLevel)
{
+ // noop
+ }
+
+ @Override
+ public void setMarkerFiler(boolean markerFile) {
+ // noop
+ }
+
+ @Override
+ public void setDeleteOrphanLockFiles(boolean deleteOrphanLockFiles) {
+ // noop
+ }
+ }
+}
diff --git
a/core/camel-core/src/test/java/org/apache/camel/impl/engine/PooledExchangeFactoryReleaseNotRoutedTest.java
b/core/camel-core/src/test/java/org/apache/camel/impl/engine/PooledExchangeFactoryReleaseNotRoutedTest.java
new file mode 100644
index 000000000000..5cb0d8ea6b51
--- /dev/null
+++
b/core/camel-core/src/test/java/org/apache/camel/impl/engine/PooledExchangeFactoryReleaseNotRoutedTest.java
@@ -0,0 +1,188 @@
+/*
+ * 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.impl.engine;
+
+import org.apache.camel.CamelContext;
+import org.apache.camel.Consumer;
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.ExtendedCamelContext;
+import org.apache.camel.PooledExchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.spi.ExchangeFactoryManager;
+import org.apache.camel.spi.PooledObjectFactory;
+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.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * A consumer that gives back an exchange it never routed (a batch cut down to
maxMessagesPerPoll, a consumer that is
+ * stopping, a failure while the message is prepared) must put it into the
pool once, so that the next two exchanges are
+ * two different objects.
+ */
+class PooledExchangeFactoryReleaseNotRoutedTest extends ContextTestSupport {
+
+ @Override
+ protected CamelContext createCamelContext() throws Exception {
+ CamelContext camelContext = super.createCamelContext();
+ ExtendedCamelContext ecc = camelContext.getCamelContextExtension();
+ ecc.setExchangeFactory(new PooledExchangeFactory());
+ ecc.getExchangeFactory().setStatisticsEnabled(true);
+ return camelContext;
+ }
+
+ @Test
+ void autoReleaseExchangeThatWasNotRoutedIsPooledOnce() {
+ Consumer consumer = context.getRoute("foo").getConsumer();
+
+ Exchange exchange = consumer.createExchange(true);
+ exchange.getMessage().setBody("not routed");
+ consumer.releaseExchange(exchange, true);
+
+ assertPooledOnce(consumer, exchange);
+ }
+
+ @Test
+ void
autoReleaseExchangeThatWasNotRoutedIsPooledOnceWhenReleasedAsNotAutoRelease() {
+ Consumer consumer = context.getRoute("foo").getConsumer();
+
+ // several consumers create their exchanges with auto release and give
back the ones they do not route with false
+ Exchange exchange = consumer.createExchange(true);
+ exchange.getMessage().setBody("not routed");
+ consumer.releaseExchange(exchange, false);
+
+ assertPooledOnce(consumer, exchange);
+ }
+
+ /**
+ * Releasing an exchange a second time is still a bug of the consumer. It
does no harm here only because nobody took
+ * the exchange from the pool in between; otherwise the second release
would reset the exchange of another message.
+ */
+ @Test
+ void autoReleaseExchangeThatWasRoutedIsPooledOnceWhenReleasedAgain()
throws Exception {
+ Consumer consumer = context.getRoute("foo").getConsumer();
+
+ Exchange exchange = consumer.createExchange(true);
+ exchange.getMessage().setBody("routed");
+ // the unit of work of the route is done here and gives the exchange
back to the pool
+ consumer.getProcessor().process(exchange);
+ assertEquals(1, statistics().getReleasedCounter());
+
+ consumer.releaseExchange(exchange, true);
+
+ assertPooledOnce(consumer, exchange);
+ }
+
+ @Test
+ void exchangeWithoutAutoReleaseIsPooledOnce() {
+ Consumer consumer = context.getRoute("foo").getConsumer();
+
+ Exchange exchange = consumer.createExchange(false);
+ exchange.getMessage().setBody("not routed");
+ consumer.releaseExchange(exchange, false);
+
+ assertPooledOnce(consumer, exchange);
+ }
+
+ @Test
+ void
releaseReturnsFalseWhenThePoolIsFullAndTheNotRoutedExchangeIsDiscarded() {
+ PooledExchangeFactory factory = newFactoryWithCapacityOne();
+
+ Exchange first = factory.create(true);
+ Exchange second = factory.create(true);
+
+ assertTrue(factory.release(first), "the pool has room for the first
exchange");
+ assertFalse(factory.release(second), "the pool is full, so the second
exchange is discarded");
+
+ assertEquals(1, factory.getSize(), "exchanges in the pool");
+ assertEquals(1, factory.getStatistics().getReleasedCounter());
+ assertEquals(1, factory.getStatistics().getDiscardedCounter());
+
+ // the discarded exchange was reset
+ assertEquals(0, second.getClock().getCreated());
+ assertSame(first, factory.create(true));
+ assertNotSame(first, factory.create(true));
+ }
+
+ @Test
+ void releaseReturnsTrueWhenTheExchangeWasAlreadyGivenBackByItsUnitOfWork()
{
+ PooledExchangeFactory factory = newFactoryWithCapacityOne();
+
+ Exchange routed = factory.create(true);
+ // what the unit of work does when the route is done
+ ((PooledExchange) routed).done();
+ assertEquals(1, factory.getSize(), "exchanges in the pool");
+
+ assertTrue(factory.release(routed), "nothing to do: the exchange is in
the pool already");
+
+ assertEquals(1, factory.getSize(), "exchanges in the pool");
+ assertEquals(1, factory.getStatistics().getReleasedCounter());
+ assertEquals(0, factory.getStatistics().getDiscardedCounter());
+ }
+
+ private PooledExchangeFactory newFactoryWithCapacityOne() {
+ PooledExchangeFactory factory = new PooledExchangeFactory();
+ factory.setCamelContext(context);
+ factory.setCapacity(1);
+ factory.setStatisticsEnabled(true);
+ // build creates the pool; the factory is not started, as it has no
consumer to be registered with
+ factory.build();
+ return factory;
+ }
+
+ private void assertPooledOnce(Consumer consumer, Exchange released) {
+ // the exchange was reset
+ assertNull(released.getMessage().getBody());
+
+ assertEquals(1, manager().getPooledCounter(), "exchanges in the pool");
+ assertEquals(1, statistics().getCreatedCounter());
+ assertEquals(1, statistics().getReleasedCounter());
+ assertEquals(0, statistics().getDiscardedCounter());
+
+ Exchange first = consumer.createExchange(true);
+ Exchange second = consumer.createExchange(true);
+
+ assertSame(released, first, "the pooled exchange is used again");
+ assertNotSame(first, second, "two messages must not share one
exchange");
+ assertEquals(0, manager().getPooledCounter(), "exchanges in the pool");
+ assertEquals(2, statistics().getCreatedCounter());
+ assertEquals(1, statistics().getAcquiredCounter());
+ }
+
+ private ExchangeFactoryManager manager() {
+ return context.getCamelContextExtension().getExchangeFactoryManager();
+ }
+
+ private PooledObjectFactory.Statistics statistics() {
+ return manager().getStatistics();
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() {
+ return new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("direct:start").routeId("foo").to("mock:result");
+ }
+ };
+ }
+}
