You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@nifi.apache.org by "Gardella Juan Pablo (Jira)" <ji...@apache.org> on 2020/01/16 14:51:00 UTC
[jira] [Updated] (NIFI-7034) Connection leak with JMSConsumer and
JMSPublisher
[ https://issues.apache.org/jira/browse/NIFI-7034?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Gardella Juan Pablo updated NIFI-7034:
--------------------------------------
Description:
JMS connections are not closed in case of a failure. Discovered against ActiveMQ, but it applies to other JMS servers.
The problem happens when an exception is raised and the worker is marked as invalid. The current code discards the worker before closing it properly. Below the details.
h3. Details
Any exception happening to a ConsumerJMS or PublisherJMS marks the worker as invalid. After that, the worker is discarded (the worker object reference is never cleaned). Below the snipped code of the issue:
{code:java|title=AbstractJMSProcessor}
} finally {
//in case of exception during worker's connection (consumer or publisher),
//an appropriate service is responsible to invalidate the worker.
//if worker is not valid anymore, don't put it back into a pool, try to rebuild it first, or discard.
//this will be helpful in a situation, when JNDI has changed, or JMS server is not available
//and reconnection is required.
if (worker == null || !worker.isValid()){
getLogger().debug("Worker is invalid. Will try re-create... ");
final JMSConnectionFactoryProviderDefinition cfProvider = context.getProperty(CF_SERVICE).asControllerService(JMSConnectionFactoryProviderDefinition.class);
try {
// Safe to cast. Method #buildTargetResource(ProcessContext context) sets only CachingConnectionFactory
CachingConnectionFactory currentCF = (CachingConnectionFactory)worker.jmsTemplate.getConnectionFactory();
cfProvider.resetConnectionFactory(currentCF.getTargetConnectionFactory());
worker = buildTargetResource(context);
}catch(Exception e) {
getLogger().error("Failed to rebuild: " + cfProvider);
worker = null;
}
}
{code}
Before discard the worker, it should be cleaned all resources associated with it. The proper solution is to call {{worker.shutdown()}} and then discard it.
was:
JMS connections are not closed in case of a failure. Discovered against ActiveMQ, but it applies to other JMS servers.
The problem happens when an exception happen and the worker is discarded and never closed properly. Below the details.
h3. Details
Any exception happening to a ConsumerJMS or PublisherJMS marks the worker as invalid. After that, the worker is discarded (the worker object reference is never cleaned). Below the snipped code of the issue:
{code:java|title=AbstractJMSProcessor}
} finally {
//in case of exception during worker's connection (consumer or publisher),
//an appropriate service is responsible to invalidate the worker.
//if worker is not valid anymore, don't put it back into a pool, try to rebuild it first, or discard.
//this will be helpful in a situation, when JNDI has changed, or JMS server is not available
//and reconnection is required.
if (worker == null || !worker.isValid()){
getLogger().debug("Worker is invalid. Will try re-create... ");
final JMSConnectionFactoryProviderDefinition cfProvider = context.getProperty(CF_SERVICE).asControllerService(JMSConnectionFactoryProviderDefinition.class);
try {
// Safe to cast. Method #buildTargetResource(ProcessContext context) sets only CachingConnectionFactory
CachingConnectionFactory currentCF = (CachingConnectionFactory)worker.jmsTemplate.getConnectionFactory();
cfProvider.resetConnectionFactory(currentCF.getTargetConnectionFactory());
worker = buildTargetResource(context);
}catch(Exception e) {
getLogger().error("Failed to rebuild: " + cfProvider);
worker = null;
}
}
{code}
Before discard the worker, it should be cleaned all resources associated with it. The proper solution is to call {{worker.shutdown()}} and then discard it.
> Connection leak with JMSConsumer and JMSPublisher
> -------------------------------------------------
>
> Key: NIFI-7034
> URL: https://issues.apache.org/jira/browse/NIFI-7034
> Project: Apache NiFi
> Issue Type: Bug
> Components: Extensions
> Environment: Discovered against ActiveMQ.
> Reporter: Gardella Juan Pablo
> Priority: Critical
> Original Estimate: 2h
> Remaining Estimate: 2h
>
> JMS connections are not closed in case of a failure. Discovered against ActiveMQ, but it applies to other JMS servers.
> The problem happens when an exception is raised and the worker is marked as invalid. The current code discards the worker before closing it properly. Below the details.
> h3. Details
> Any exception happening to a ConsumerJMS or PublisherJMS marks the worker as invalid. After that, the worker is discarded (the worker object reference is never cleaned). Below the snipped code of the issue:
> {code:java|title=AbstractJMSProcessor}
> } finally {
> //in case of exception during worker's connection (consumer or publisher),
> //an appropriate service is responsible to invalidate the worker.
> //if worker is not valid anymore, don't put it back into a pool, try to rebuild it first, or discard.
> //this will be helpful in a situation, when JNDI has changed, or JMS server is not available
> //and reconnection is required.
> if (worker == null || !worker.isValid()){
> getLogger().debug("Worker is invalid. Will try re-create... ");
> final JMSConnectionFactoryProviderDefinition cfProvider = context.getProperty(CF_SERVICE).asControllerService(JMSConnectionFactoryProviderDefinition.class);
> try {
> // Safe to cast. Method #buildTargetResource(ProcessContext context) sets only CachingConnectionFactory
> CachingConnectionFactory currentCF = (CachingConnectionFactory)worker.jmsTemplate.getConnectionFactory();
> cfProvider.resetConnectionFactory(currentCF.getTargetConnectionFactory());
> worker = buildTargetResource(context);
> }catch(Exception e) {
> getLogger().error("Failed to rebuild: " + cfProvider);
> worker = null;
> }
> }
> {code}
> Before discard the worker, it should be cleaned all resources associated with it. The proper solution is to call {{worker.shutdown()}} and then discard it.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)