Federico Mariani created CAMEL-26658:
----------------------------------------
Summary: camel-amqp - the producer opens a new broker connection
per message when the component builds the ConnectionFactory
Key: CAMEL-26658
URL: https://issues.apache.org/jira/browse/CAMEL-26658
Project: Camel
Issue Type: Improvement
Components: camel-amqp
Affects Versions: 4.18.5, 4.23.0, 4.22.2
Reporter: Federico Mariani
h3. Problem
When {{camel-amqp}} builds the {{ConnectionFactory}} itself, from {{brokerUrl}}
or from {{host}}/{{port}}/{{username}}/{{password}}/{{useSsl}}, the producer
opens a new AMQP connection for every message and closes it again. On a real
network that is TCP, TLS, SASL and AMQP open/begin/attach per send.
*
{{components/camel-amqp/src/main/java/org/apache/camel/component/amqp/AMQPComponent.java}}
lines 107-162 ({{doInit}}) set a plain qpid-jms {{JmsConnectionFactory}} on
the configuration. The static factories {{AMQPComponent.amqpComponent(...)}}
(lines 92-102) do the same.
* {{camel-jms}} does not wrap it:
** {{JmsConfiguration.createInOnlyTemplate}} (line 780; also used by the InOut
template) takes {{getOrCreateTemplateConnectionFactory()}}.
** That falls back to the same factory ({{createTemplateConnectionFactory}},
lines 1875-1877).
* {{JmsProducer.doSend}} (lines 453-463) calls {{CamelJmsTemplate.send}}, which
calls Spring {{JmsTemplate.execute(callback, false)}}. With no transactional
session bound, it calls {{createConnection()}} and closes the connection after
each call (spring-framework 7.0.10 {{JmsTemplate.java}} lines 520-548; the
class javadoc asks for a caching or pooling factory).
* {{cacheLevel}} only applies to the listener containers, not to the template.
* {{camel-activemq}} wraps the factory it builds in a pool by default
({{usePooledConnection=true}}), and the {{camel-jms}} docs explain pooling. The
{{camel-amqp}} docs never mention it, and the YAML/properties example at
{{amqp-component.adoc}} line 112 binds a plain {{JmsConnectionFactory}}.
h3. How to reproduce
Measured on main 7978db3d42dd, with a test in {{camel-amqp}}:
* Embedded Artemis 2.57.0 with the AMQP acceptor
({{ArtemisServiceFactory.createAMQPService()}}).
* Route {{from("direct:send").to("amqp:queue:verify")}}: 50 warm-up sends, then
30 measured.
* Broker connections are counted with
{{ActiveMQServer.getTotalConnectionCount()}}.
* Loopback, no TLS or SASL.
||Component setup||Factory used||New broker connections for 30 sends||Median
per send||
|{{setHost("localhost")}}, {{setPort(port)}}|{{JmsConnectionFactory}}|30|5.4 ms|
|{{setBrokerUrl("amqp://localhost:port")}}|{{JmsConnectionFactory}}|30|4.6 ms|
|control: {{setConnectionFactory(new
CachingConnectionFactory(plain))}}|{{CachingConnectionFactory}}|0|0.55 ms|
A counting proxy around the factory the component built recorded 30
{{createConnection()}} calls for 30 sends.
h3. Proposed change
* Wrap only the template connection factory that the component builds
({{setTemplateConnectionFactory}}) in Spring's {{CachingConnectionFactory}}.
** It is already on the classpath through {{spring-jms}}, so there is no new
dependency. pooled-jms, which {{camel-activemq}} uses, is not a {{camel-amqp}}
dependency.
** Put it behind a new component option (for example {{cacheConnection}}) that
defaults to {{true}}.
** Apply it also to the static {{amqpComponent(...)}} factories.
** Leave a factory supplied by the user untouched.
* Destroy the caching factory (close its shared connection) when the component
stops.
* Keep the listener containers on the plain factory. The Spring
{{DefaultMessageListenerContainer}} javadoc lists the limitations of a
{{CachingConnectionFactory}} under a listener container.
* *Transactions (source reading, not tested):*
** The lazily created {{JmsTransactionManager}} uses
{{getOrCreateConnectionFactory()}} ({{JmsConfiguration}} line 1884).
** Spring looks up the transactional session by factory instance
({{ConnectionFactoryUtils}} line 286).
** So a producer with a wrapped template factory would no longer join that
transaction. Skip the wrapping when {{transacted=true}} or a
{{transactionManager}} is set, or build the transaction manager on the same
factory.
* {{CachingConnectionFactory}} caches 1 session by default
({{sessionCacheSize=1}}). Concurrent producers beyond that still create a
session per send on the shared connection; consider a larger size or exposing
it.
* *Docs:*
** Change the example at {{amqp-component.adoc}} line 112 to
{{camel.component.amqp.broker-url = amqp://localhost:5672}}.
** Add a short section on connection caching and pooling for factories supplied
by the user, as {{jms-component.adoc}} has.
h3. Impact / upgrade
* This changes a default: one shared connection instead of one per message, for
the factories the component builds. It needs an entry in
{{camel-4x-upgrade-guide-4_24.adoc}}, including how to switch it off.
* Request/reply producers use the same template factory and are covered.
* Consumers are not affected.
* The same code is in 4.18.5, 4.22.2 and 4.23.0.
h3. Related
* CAMEL-22161: added the {{brokerUrl}} component option.
* CAMEL-20521: transacted sends.
* CAMEL-12070: a {{CachingConnectionFactory}} that was not shut down with the
bundle. The cache must be stopped with the component.
_Claude Code on behalf of Federico Mariani_
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
