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 28e150782cdc CAMEL-25559: camel-core - PooledExchangeFactory puts an
exchange that was given back without being routed into the pool twice (#27698)
28e150782cdc is described below
commit 28e150782cdc96cdaf0e635639c08d8aefc50306
Author: allthingssecurity <[email protected]>
AuthorDate: Mon Oct 12 23:59:29 2026 +0530
CAMEL-25559: camel-core - PooledExchangeFactory puts an exchange that was
given back without being routed into the pool twice (#27698)
With the pooled exchange factory (camel.main.exchangeFactory=pooled), a
consumer that gave back an exchange it did not route put it into the pool
twice: release() called done(), whose on-done task called release() again,
and
both calls offered the exchange. The next two messages of the consumer then
shared one exchange object. This hit GenericFileConsumer (files whose read
lock
was not acquired, files beyond maxMessagesPerPoll), the mail, cloud storage
and
other consumers that give back unrouted exchanges.
PooledExchangeFactory.release now offers an auto-release exchange to the
pool
exactly once: the on-done task offers it directly on the routed path, a
release
of an already done exchange does nothing, and a release of an exchange not
yet
routed suspends the task for its own done() and offers the exchange itself,
so
it returns false when the pool is full. Every caller is covered without
changing any consumer.
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 7cb87af1592e..a264fbece4cd 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
@@ -96,13 +96,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);
@@ -115,12 +151,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) {
@@ -160,7 +200,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");
+ }
+ };
+ }
+}
