The AbstractJmsTransformer was opening a JMS session but not closing it. We should also look into an alternative to creating a JMS session just in the transformer (to reduce overhead)

Fix committed: r6970

So, help me out here. I could do something similar to the JmsMessageDispatcher where I only close the session if it's not in a transacted or cached?

It's too bad we have to open a JMS session at all in the AbstractJmsTransformer at all, but I guess that's the only way to create the JmsMessage ?

I checked in a fix, which is similar to my previous fix but handles the case where the session in a transaction. All jms tests passed, including the transaction tests.

This fix is for a customer case (00001091). I checked the fix into the sandbox this time, since it's a 1.4.0 fix that needs to be merged to trunk.

r7031

Here is my suggested fix, I have tested with both no transaction and resource local transaction (haven't tested with XA yet but I will). I'd be happy to try out your proposed fix, if you can post it here (I don't have subversion)?

protected Message transformToMessage(Object src) throws TransformerException {

Session session = null;
try {
Message result;

if (src instanceof Message) { result = (Message) src; result.clearProperties(); } else { session = this.getSession(); result = JmsMessageUtils.toMessage(src, session); }

// set the event properties on the Message
UMOEventContext ctx = RequestContext.getEventContext();
if (ctx == null) { logger.warn("There is no current event context"); return result; }

this.setJmsProperties(ctx.getMessage(), result);

return result;
} catch (TransformerException tex) { // rethrow throw tex; } catch (Exception e) { throw new TransformerException(this, e); } finally {
if (session != null) {
UMOTransaction tx = TransactionCoordination.getInstance().getTransaction();
JmsConnector conn = (JmsConnector)endpoint.getConnector();
if (tx == null || !tx.hasResource(conn.getConnection())) {
try { logger.info("closing JMS session"); session.close(); } catch (JMSException e) { logger.info("Exception closing JMS session: " +e.getMessage(), e); }
}
}
}
}

Justin,

Yes, looks like we have roughly the same fix, except I just used a slightly different criteria for whether the session is in a transcation. (Your criteria may actually be better).

Here's the fix I proposed:

http://svn.codehaus.org/mule/trunk/mule-sandbox/transports/jms-leak-1.4.0/

I also had to modify one of the test cases slightly, that's nested down in the url above.

If you could test, that would be great. I will do the same and test your fix.

Thanks!
quoc

BTW session caching is performed in the message dispatcher, as such it is not available in the jms connector, so unless the jms connector is reworked to implement session caching you can't access the cached session from the transformer. Session caching is only applicable if there is no transaction available.

Another interesting thing I discovered is that the jms receiver endpoint also invokes the ObjectToJmsMessage response transformer by default, even if you set synchronous="false" on the endpoint, seems to be redundant in that case?

Cheers,

Justin

Sounds good, yes I will merge the change to trunk and hopefully it gets into 1.4.2.

I appreciate the help and insight, Justin. That's great to hear the WebSphere tests also pass. Btw, the customer also verified the fix in their WebLogic environment.

1. Event from Transacted JMS Receiver -> JMS Transformer -> JMS Dispatch (one or n-times)
Here JMS session is always available through JMSConnector.getSessionFromTransaction. And the design is clear: TransactedJmsMessageReceiver class keeps whole transaction processing encapsulated. It knows when to start, keep the session, and end transaction. So the transformer gets JMS session from transaction context.

2. Event from somewhere else -> JMS Transformer -> JMS Dispatch (one or n-times) without transaction
First JMS Session is created in Transformer and then the second one during dispatch in JMSMessageDispatcher. It can be cached within endpoint. When there's a dynamic endpoint from AbstractRecipientList where there's no control over endpoint cache, session count starts to grow depending on number of cached enpoints.

3. Event from somewhere else -> JMS Transformer -> JMS Dispatch (one or n-times) within transaction.
JMS Session is created in Transformer, because it is first time when session is needed. It's bound do transaction ctx but after commit/rollback this session is lost. When next event comes to transformer it creates another transaction with a new JMS session.

I felt a bit lost, so tried to find solution somewhere else. So I looked at JMSConnector - and I think this class is right the place for storing JMS sessions. They should be kept like JDBC Connections in the managed pool. The number of sessions needed is a count of concurrent JMS dispatchers and probably depends on threading profile of Mule components. I tried to do quite primitive pool that controls creation of JMS sessions. Here's the code. Better implementations could decorate JmsTransaction class to release session to the pool.

/**

• Gets transformed message from object while cleanly closing any sessions Added per MULE-1894
• @param src
• source object
• @return <code>Message</code> transformed message

• */

protected final Message getResult(Object src) throws JMSException, TransformerException {

boolean transacted = false;
boolean cached = false;

// Figure out if the session comes from a Transaction or Cache. This
// affects how we close the session in this transformer.
// Note: This session is only used for

// TODO: JmsMessageDispatcher has similar code to figure out if session is in transaction

Session session = null;

JmsConnector connector = (JmsConnector) endpoint.getConnector();

session = connector.getSessionFromTransaction();

if (session != null) { transacted = true; } else {

session = this.getSession();

UMOTransaction tx = TransactionCoordination.getInstance().getTransaction();
if ((tx != null) && (tx.hasResource(connector.getConnection()))) {
transacted = true;

}

}

// TODO: figure out if JmsSession caching is on. Since it's off by default,
// deferring this for now.

try { return JmsMessageUtils.toMessage(src, session); } finally {

if (session != null && !transacted && !cached) { session.close(); }

}

}

The JMSMessageDispatcher checks the endpoint configuration to determine if it is transacted. This means that it is neccessary to add an ALWAYS_JOIN transaction on the jms endpoint, or else it wont participate in the transaction. The solution is to check if the connector has been registered in the transaction (by connector.getSession), then it will be automatically enlisted without need to add a transaction on the endpoint. See the underlined code below...

private UMOMessage dispatchMessage(UMOEvent event) throws Exception
{
Session session = null;
MessageProducer producer = null;
MessageConsumer consumer = null;
Destination replyTo = null;
boolean transacted = false;
boolean cached = false;
boolean remoteSync = useRemoteSync(event);

if (logger.isDebugEnabled())

try
{
session = connector.getSessionFromTransaction();
if (session != null)
{
transacted = true;

// If a transaction is running, we can not receive any messages
// in the same transaction.
if (remoteSync)

}
// Should we be caching sessions? Note this is not part of the JMS spec.
// and is turned off by default.
else if (event.getMessage().getBooleanProperty(JmsConstants.CACHE_JMS_SESSIONS_PROPERTY,
connector.isCacheJmsSessions()))
{
cached = true;
if (cachedSession != null)

else

}
else
{
session = connector.getSession(event.getEndpoint());
// check global transaction, not endpoint transaction config
// since endpoint transaction config need not be set
UMOTransaction tx = TransactionCoordination.getInstance().getTransaction();
if ((tx != null) && (tx.hasResource(connector.getConnection()))) {
transacted = true;
}

/* if (event.getEndpoint().getTransactionConfig().isTransacted())

*/
}

Where do we stand with this issue. How long would it take to finish up the work?

Thank you,
-Andreas

Incorporating suggestions from Leon and Justin to address setting of the transacted flag under certain scenarios, I'd say there's 3-4 days left to analyze, test, and check in. However, I am completely swamped.

Thanks,
quoc

Andrew Calafato and I have also stumbled across this problem. Seeing that you're swamped do you need a hand?

Thanks
Marie

I made some changes and put the session on the thread, to avoid opening and closing of sessions in the transformer. The session leak seems to be fixed. However, I'm having problems with the xa transactions not picking up all the messages on the queue. I'm not sure whether this is related so I'm not committing anything at this time.

http://fisheye.codehaus.org/changelog/mule/?cs=9693

Comments inlined in the changeset