Regression: Using static queue as a reply queue in InOut pattern causes memory leak [CAMEL-2740]

When using camel-jms with pattern InOut and a fixed/shared replyTo address and a clustered activemq broker, a steady increase in broker memory usage is observed. Looking into the heap, we can see that the bulk of the memory is accumulated in an instance of org.apache.activemq.broker.region.cursors.VMPendingMessageCursor and is consumed by AdvisoryTopic subscriptions. Digging further into the heap, we can see that the bulk of the messages in the pending queue are related to consumer additions and deletions:

select * from org.apache.activemq.command.ActiveMQMessage msg where toString(msg.destination.physicalName).contains("ActiveMQ.Advisory.Consumer")

Digging a little further, we can see also similar symptoms to the previous CAMEL-2740 issue: a slow accumulation of ConsumerInfo objects for the reply queues with selector:

JMSCorrelationID='CamelDummyJmsMessageSelector'

select * from org.apache.activemq.command.ConsumerInfo ci where toString(ci.selector).contains("CamelDummyJmsMessageSelector")

One heap I examined had an accumulation of over 500000 consumer-related AdvisoryTopic messages with 260000 ConsumerInfo objects for replyTo consumers with the above selector expression.

When using shared reply queues in a network of brokers, the churn in consumer updates (to update selectors) also results in a build up of orphaned ConsumerInfo objects eventually causing the brokers in the network to run out of heap space.

This is observed when a replyToCacheLevelName of CACHE_SESSION or CACHE_NONE is set. When CACHE_CONSUMER is used, an increase in consumers per reply destination is not observed; however, dispatch is blocked by consumers with selector expression JMSCorrelationID='CamelDummyJmsMessageSelector'