messaginghub / pooled-jms Goto Github PK
View Code? Open in Web Editor NEWA JMS Connection pool for messaging applications supporting JMS 1.1 and 2.0 and Jakarta JMS clients
License: Apache License 2.0
A JMS Connection pool for messaging applications supporting JMS 1.1 and 2.0 and Jakarta JMS clients
License: Apache License 2.0
If using spring boot and wrapping a connection factory as a bean that is referenced in a camel context the PooledConnection ends up destroying the provided exceptionListener.
@Bean
public ConnectionFactory consumerConnectionFactory(ActiveMqProperties properties) {
JmsPoolConnectionFactory poolConnectionFactory = new JmsPoolConnectionFactory();
JmsConnectionFactory jmsConnectionFactory = new JmsConnectionFactory(
properties.getConsumer().getUsername(),
properties.getConsumer().getPassword(),
properties.getConsumer().getBrokerUrl()
);
poolConnectionFactory.setConnectionFactory(jmsConnectionFactory);
// This will never run because its overwritten when the PooledConnection is instantiated:
ExceptionListener exceptionListener = new ExceptionListener() {
@Override
public void onException(JMSException arg0) {
System.out.println("Sample" + arg0.getMessage());
}
};
jmsConnectionFactory.setExceptionListener(exceptionListener);
return poolConnectionFactory;
}
You can fix this erasure by preserving the original exception listener in the PooledConnection constructor:
public PooledConnection(Connection connection) {
final GenericKeyedObjectPoolConfig poolConfig = new GenericKeyedObjectPoolConfig<>();
poolConfig.setJmxEnabled(false);
this.connection = wrap(connection);
try {
this.setParentExceptionListener(connection.getExceptionListener())
this.connection.setExceptionListener(this);
Otherwise the connections have to be taken out of the pool and updated manually with an exception handler inbetween spring bean instantiation and camel route configuration.
https://github.com/messaginghub/pooled-jms/blob/master/pom.xml#L78
Using JDK 11 on Ubuntu Bionic, I get a build failure due to a too-old source version. The internet says the culprit is maven-compiler-plugin < 3.8.0. (Option 2 at https://stackoverflow.com/questions/49028810/error-source-option-1-5-is-no-longer-supported-use-1-6-or-later)
[ERROR] COMPILATION ERROR :
[INFO] -------------------------------------------------------------
[ERROR] Source option 5 is no longer supported. Use 6 or later.
[ERROR] Target option 1.5 is no longer supported. Use 1.6 or later.
[INFO] 2 errors
[INFO] -------------------------------------------------------------
[INFO] ------------------------------------------------------------------------
[INFO] BUILD FAILURE
[INFO] ------------------------------------------------------------------------
[INFO] Total time: 8.842 s
[INFO] Finished at: 2019-05-24T19:48:33Z
[INFO] ------------------------------------------------------------------------
[ERROR] Failed to execute goal org.apache.maven.plugins:maven-compiler-plugin:3.1:compile (default-compile) on project messaging-examples: Compilation failure: Compilation failure:
[ERROR] Source option 5 is no longer supported. Use 6 or later.
[ERROR] Target option 1.5 is no longer supported. Use 1.6 or later.
[ERROR] -> [Help 1]
In an attempt to migrate Spring Boot auto-configurations from activemq-pool
to this project, I've realized a few properties where removed (apparently in this commit that doesn't reference an issue so can't comment there sorry.
We're still considering migrating to this project for the JMS 2.0 API compatibility but it would be nice to get some more information as why those properties were removed so that we can properly deprecate them.
Thanks!
Actually no value for timeBetweenEvictionRunsMillis is settable, thus forcing to disable evictor thread.
The logic within org.apache.commons.pool2.impl.GenericKeyedObjectPool.evict()
gives the EvictionPolicy
higher precedence than KeyedPooledObjectFactory.validateObject()
.
This means the comment in org.messaginghub.pooled.jms.JmsPoolConnectionFactory.initConnectionsPool()
is not true
// We always want our validate method to control when idle objects are evicted.
this.connectionsPool.setTestOnBorrow(true);
this.connectionsPool.setTestWhileIdle(true);
Since JmsPoolConnectionFactory
leaves all pooled connections with a state of PooledObjectState.IDLE
, they will be evicted after BaseObjectPoolConfig.DEFAULT_MIN_EVICTABLE_IDLE_DURATION
, 30 minutes by default.
This causes code that is using an active connection to stacktrace, example:
javax.jms.IllegalStateException: The Session is closed
at org.apache.activemq.ActiveMQSession.checkClosed(ActiveMQSession.java:772) ~[activemq-client-5.16.5.jar:5.16.5]
at org.apache.activemq.ActiveMQSession.rollback(ActiveMQSession.java:597) ~[activemq-client-5.16.5.jar:5.16.5]
at org.messaginghub.pooled.jms.JmsPoolSession.close(JmsPoolSession.java:112) ~[pooled-jms-1.2.4.jar:na]
In PooledConnectionFactoryTest
:
@Test(timeout = 60000)
public void testMaxConnectionsAreExceeded() throws Exception {
JmsPoolConnectionFactory cf = createPooledConnectionFactory();
cf.setMaxConnections(2);
JmsPoolConnection conn1 = (JmsPoolConnection) cf.createConnection("admin", "admin");
conn1.start();
JmsPoolConnection conn2 = (JmsPoolConnection) cf.createConnection("user", "password");
conn2.start();
JmsPoolConnection conn3 = (JmsPoolConnection) cf.createConnection("guest", "password");
conn3.start();
assertNotSame(conn1.getConnection(), conn2.getConnection());
assertNotSame(conn1.getConnection(), conn3.getConnection());
assertNotSame(conn2.getConnection(), conn3.getConnection());
assertEquals(2, cf.getNumConnections()); // this throws: expected 2, got 3
cf.stop();
}
Using the latest tag 1.0.0, we get this error when executing the following code:
JmsPoolConnectionFactory poolingFactory = new JmsPoolConnectionFactory();
poolingFactory.setConnectionFactory(createArtemisConnectionFactory());
TopicConnection topicConnection = poolingFactory.createTopicConnection();
TopicSession topicSession = topicConnection.createTopicSession(false, Session.AUTO_ACKNOWLEDGE);
Topic topic = ic.lookup("topicname");
TopicPublisher topicPublisher = topicSession.createPublisher (topic); <- throws exception
java.lang.NoSuchMethodError: javax.jms.MessageProducer.getDeliveryDelay()J
at org.messaginghub.pooled.jms.JmsPoolMessageProducer.(JmsPoolMessageProducer.java:64)
at org.messaginghub.pooled.jms.pool.PooledSessionHolder.getOrCreateProducer(PooledSessionHolder.java:144)
at org.messaginghub.pooled.jms.JmsPoolSession.createProducer(JmsPoolSession.java:426)
We are using Artemis version 2.4.0.
Configurations.md describes options createConnectionOnStartup
, expiryTimeout
, and reconnectOnException
, which seem to have been removed or renamed.
Besides these, I figure , timeBetweenExpirationCheckMillis
was renamed to connectionCheckInterval
.
And maximumActiveSessionPerConnection
was renamed to maxSessionsPerConnection
.
Just wondering if this library will be updated to use jakarta.jms instead of javax.jms. If that's the case, is there a timeline?
Thanks
We are running pooled-jms in kubernetes and it would be useful to have jms pool metrics (e.g. number of active / idle connections and sessions per pool) so we can monitor what's going on in our pods.
I noticed that micrometer by default already exposes all pool metrics, but they expect that jmx is enabled.
So I think that a feature to enable jmx on the pool would work for us as well.
2018-05-10 15:16:16,440 [main ] INFO JmsPoolConnectionFactory - Provided ConnectionFactory is JMS 2.0+ capable.
2018-05-10 15:16:16,629 [me-limited test] INFO JmsPoolReconnectOnFailureTest - Fetched new connection from the pool: JmsPoolConnection { ConnectionPool[org.messaginghub.pooled.jms.mock.MockJMSConnection@720d4b3] }
2018-05-10 15:16:16,716 [5e-f06f291b95df] INFO JmsPoolReconnectOnFailureTest - Pooled Connection failed
2018-05-10 15:16:16,721 [main ] DEBUG JmsPoolConnectionFactory - Stopping the PooledConnectionFactory, number of connections in cache: 0
2018-05-10 15:16:16,721 [main ] INFO JmsPoolTestSupport - ========== finished test testConnectionCanBeCreatedAfterFailure ==========
java.lang.AssertionError: Should be disconnected
at org.junit.Assert.fail(Assert.java:88)
at org.messaginghub.pooled.jms.JmsPoolReconnectOnFailureTest.testConnectionCanBeCreatedAfterFailure(JmsPoolReconnectOnFailureTest.java:87)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.base/java.lang.reflect.Method.invoke(Method.java:564)
at org.junit.runners.model.FrameworkMethod$1.runReflectiveCall(FrameworkMethod.java:50)
at org.junit.internal.runners.model.ReflectiveCallable.run(ReflectiveCallable.java:12)
at org.junit.runners.model.FrameworkMethod.invokeExplosively(FrameworkMethod.java:47)
at org.junit.internal.runners.statements.InvokeMethod.evaluate(InvokeMethod.java:17)
at org.junit.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:298)
at org.junit.internal.runners.statements.FailOnTimeout$CallableStatement.call(FailOnTimeout.java:292)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
at java.base/java.lang.Thread.run(Thread.java:844)
I've got following exception from pooled-jms code:
java.util.NoSuchElementException: Pool exhausted
at org.apache.commons.pool2.impl.GenericKeyedObjectPool.borrowObject(GenericKeyedObjectPool.java:362)
at org.apache.commons.pool2.impl.GenericKeyedObjectPool.borrowObject(GenericKeyedObjectPool.java:265)
at org.messaginghub.pooled.jms.JmsPoolConnectionFactory.createJmsPoolConnection(JmsPoolConnectionFactory.java:721)
... 109 common frames omitted
I'm using version 1.0.5 and have configured connectionIdleTimeout/connectionCheckInterval properties on my JmsPoolConnectionFactory.
I suspect that wrong exception is caught in JmsPoolConnectionFactory.createJmsPoolConnection method in line 772.
In the class org.messaginghub.pooled.jms.pool.PooledXAConnection, the method enlistResource of TransactionManager is invoked, but the return value of this method is ignored. The return value is a boolean; accoriding to the spec, it is true if the resource was enlisted, and false otherwise.
We are using the PooledXAConnection in combination with the Narayana transaction manager (org.jboss.narayana.jta:narayana-jta:5.12.7.Final). This implementation sometimes returns false for enlistResource if a resource is not available (in such a case, the transaction is marked as rollback-only). Since this return value is not checked, the processing continues despite the absence of a working transaction. In our case, this has lead to situations where messages could not be processed and were lost.
It would be good to check the return value of enlistResource, and perform appropriate actions if the return value is false.
Stack trace, in case it can be useful:
ARJUNA016061: TransactionImple.enlistResource - XAResource.start returned: XAException.XAER_RMFAIL for < formatId=131077, gtrid_length=50, bqual_length=36, tx_uid=0:ffffac1edcbf:-3c0b9c4d:651eb4d8:225e2, node_name=grnapj0080linprdduonl8, branch_uid=0:ffffac1edcbf:-3c0b9c4d:651eb4d8:225e6, subordinatenodename=null, eis_name=unknown eis name >: javax.transaction.xa.XAException: The method 'xa_start' has failed with errorCode '-7'.
at com.ibm.mq.jmqi.JmqiXAResource.start(JmqiXAResource.java:1019)
at com.arjuna.ats.internal.jta.transaction.arjunacore.TransactionImple.enlistResource(TransactionImple.java:661)
at com.arjuna.ats.internal.jta.transaction.arjunacore.TransactionImple.enlistResource(TransactionImple.java:422)
at org.messaginghub.pooled.jms.pool.PooledXAConnection.createSession(PooledXAConnection.java:72)
at org.messaginghub.pooled.jms.JmsPoolConnection.createSession(JmsPoolConnection.java:194)
at org.springframework.jms.support.JmsAccessor.createSession(JmsAccessor.java:214)
at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.access$200(AbstractPollingMessageListenerContainer.java:78)
at org.springframework.jms.listener.AbstractPollingMessageListenerContainer$MessageListenerContainerResourceFactory.createSession(AbstractPollingMessageListenerContainer.java:509)
at org.springframework.jms.connection.ConnectionFactoryUtils.doGetTransactionalSession(ConnectionFactoryUtils.java:328)
at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.doReceiveAndExecute(AbstractPollingMessageListenerContainer.java:294)
at org.springframework.jms.listener.AbstractPollingMessageListenerContainer.receiveAndExecute(AbstractPollingMessageListenerContainer.java:246)
at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.invokeListener(DefaultMessageListenerContainer.java:1237)
at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.executeOngoingLoop(DefaultMessageListenerContainer.java:1227)
at org.springframework.jms.listener.DefaultMessageListenerContainer$AsyncMessageListenerInvoker.run(DefaultMessageListenerContainer.java:1120)
at java.lang.Thread.run(Thread.java:750)
I've seen the following javadoc in the code:
/**
* Determines if this Connection has expired.
* <p>
* A ConnectionPool is considered expired when all references to it are released AND either
* the configured idleTimeout has elapsed OR the configured expiryTimeout has elapsed.
* Once a ConnectionPool is determined to have expired its underlying Connection is closed.
*
* @return true if this connection has expired.
*/
public synchronized boolean expiredCheck() {
I don't see any way to configure an expiryTimeout. I need my connection to not live more than for example 15 min, is that possible?
If the JMS provider supports allowing the remote to close resources such as session and producers the cached producer instances can be held forever and never again successfully send a message. Some minimal effort should be made to detect that a resource was closed and evict it from the pool or cleared from session cache so new instance can be recreated once an exception is thrown.
When trying to reconnect the connection after idle timeout, we are receiving following error. Can you help me to find out what the problem is? MessageBroker is ActiveMQ Artemis
2022-03-15 20:57:39.949 ERROR 1184 --- [Camel (camel-1) thread #26 - JmsConsumer[agrar.sfm.cloud.salesinvoice.invoices.import.internal]] c.c.j.DefaultJmsMessageListenerContainer : Could not refresh JMS Connection for destination 'agrar.sfm.cloud.salesinvoice.invoices.import.internal' - retrying using FixedBackOff{interval=5000, currentAttempts=0, maxAttempts=unlimited}. Cause: Error while attempting to add new Connection to the pool
I've got following exception (same as reported in #15) from the latest pooled-jms code (3.1.0) as well as from version 1.2.4:
java.util.NoSuchElementException: Pool exhausted
at org.apache.commons.pool2.impl.GenericKeyedObjectPool.borrowObject(GenericKeyedObjectPool.java:457)
at org.apache.commons.pool2.impl.GenericKeyedObjectPool.borrowObject(GenericKeyedObjectPool.java:350)
at org.messaginghub.pooled.jms.JmsPoolConnectionFactory.createJmsPoolConnection(JmsPoolConnectionFactory.java:716)
or from version 1.2.4
at org.messaginghub.pooled.jms.JmsPoolConnectionFactory.createJmsPoolConnection(JmsPoolConnectionFactory.java:732)
Please use the same test method to reproduce the issue as reported in previous related issue - it should be added to org.messaginghub.pooled.jms.JmsPoolConnectionFactoryTest, optionally changing 1ms to 2ms as below because in case of 1ms it sometimes fails somewhere else. This is random issue so sometimes this tests fails sometimes it passes.
@Test
public void testCreateConnectionWithIdleTimeoutAndCheckInterval() throws Exception {
cf.setConnectionIdleTimeout(2);
cf.setConnectionCheckInterval(2);
for (int i = 0;i< 10000;++i){
Connection connection = cf.createConnection();
assertNotNull(connection);
connection.close();
Thread.sleep(2);
}
}
When an error occurs during the registering of the tx-sync in PooledXAConnection.createSession, the session is not put back in the pool and leaks.
Caused by: javax.transaction.RollbackException: ARJUNA016083: Cannot register synchronization because the transaction is in aborted state
at com.arjuna.ats.internal.jta.transaction.arjunacore.TransactionImple.registerSynchronizationImple(TransactionImple.java:395)
at com.arjuna.ats.internal.jta.transaction.arjunacore.TransactionImple.registerSynchronization(TransactionImple.java:376)
at org.messaginghub.pooled.jms.pool.PooledXAConnection.createSession(PooledXAConnection.java:70)
Find a small sample project: https://github.com/apinske/playground-mq/tree/narayana-leak
Hello!
This is no(t yet an) issue, but more of a question.
We are using:
quarkus
(latest LTS, 3.8.3
) withcamel-jms
(latest LTS, 4.4.1
, which - under the hood - uses spring
's JmsTemplate
in version 6.1.6
)to write multiple messages to a queue within an XA transaction. Our backing messaging system is an ActiveMQ Artemis. As driver, we use
quarkus-artemis-jms
in version 3.2.1
(which - under the hood - uses artemis
in version 3.32.0
) andquarkus-pooled-jms
in version 2.3.1
(which - under the hood - uses messaginghub
's pooled-jms
version 3.1.5
) to pool the connections.We had an issue where the pool started to block after 500 messages have been sent in a singular XA transaction. We were able to prevent this issue by setting quarkus.pooled-jms.max-sessions-per-connection
(which defaults to 500
) to -1
(which is then used to configure the maxSessionsPerConnection
on the JmsPoolConnectionFactory
here (github.com
)).
This indicates that the pool is opening a separate session per message sent in the XA transaction.
We wrote a reproducer which can be found here (github.com
). We can execute
./mvnw clean quarkus:dev
to start the reproducer. It will write 1000 messages to the queue on application startup.
To run the reproducer:
docker
or a docker
-compatible container runtime - like podman
- as well astestcontainers
setupis needed.
JmsTemplate
(docs.spring.io
) we find the following:maxSessionsPerConnection
to -1
? Especially:
... or even implement AutoCloseable.
When a pooled consumer is closed the pooled session is removing the wrong reference from its internal tracking leading to a leak whereby it retains a reference to the closed pooled consumer wrapper. The code needs to be tightened to avoid allowing this to happen and ensure that the pooled wrappers are removed properly.
A declarative, efficient, and flexible JavaScript library for building user interfaces.
๐ Vue.js is a progressive, incrementally-adoptable JavaScript framework for building UI on the web.
TypeScript is a superset of JavaScript that compiles to clean JavaScript output.
An Open Source Machine Learning Framework for Everyone
The Web framework for perfectionists with deadlines.
A PHP framework for web artisans
Bring data to life with SVG, Canvas and HTML. ๐๐๐
JavaScript (JS) is a lightweight interpreted programming language with first-class functions.
Some thing interesting about web. New door for the world.
A server is a program made to process requests and deliver data to clients.
Machine learning is a way of modeling and interpreting data that allows a piece of software to respond intelligently.
Some thing interesting about visualization, use data art
Some thing interesting about game, make everyone happy.
We are working to build community through open source technology. NB: members must have two-factor auth.
Open source projects and samples from Microsoft.
Google โค๏ธ Open Source for everyone.
Alibaba Open Source for everyone
Data-Driven Documents codes.
China tencent open source team.