Apache Camel RabbitMQ leaving behind threads in WAIT state

Question

I have a set of Camel routes configured to read and write to RabbitMQ queues, more or less like this:

from("rabbitmq:$rabbitMQVhost?connectionFactory=#customConnectionFactory&queue=${it.rabbitMQQueue}&routingKey=${it.rabbitMQQueue}&SOME_MORE_PROPERTIES")
    .log("Read message from queue ${it.rabbitMQQueue}")
    .routeId(it.rabbitMQQueue)
    .noAutoStartup()
    .bean(it.rabbitMQBean)
    .choice()
    .`when`(PredicateBuilder.and(simple("$myCondition"), isNotNull(body())))
        .split(body())
        .toD("rabbitmq:$rabbitMQVhost?connectionFactory=#customConnectionFactory&queue=${it.rabbitMQQueueDestination}&autoDelete=false&routingKey=${it.rabbitMQQueueDestination}&bridgeEndpoint=true")
        .endChoice()
    .otherwise()
    end()

Where SOME_MORE_PROPERTIES is basically autoDelete=false&autoAck=false and some message prefetch settings.

My ConnectionFactory is a org.springframework.amqp.rabbit.connection.CachingConnectionFactory.

Whenever a message comes in on my source queue, a thread is started to process it; however, after the processing is completed it hangs in WAIT state, never being released or terminated, so my application memory saturates after a while and there's nothing the garbage collector can do about it.

After some time running, my application is basically in this state:

If I manually restart the routes, the threads are terminated and the memory released.

Is there something I'm doing wrong in my routes configuration that is preventing the threads from terminating properly?

I'd like to avoid having to write a quartz job to restart the routes every once in a while.

Edit: I also recently updated from Camel 2.24.0 to the latest RC for Camel 3, but the issue is still happening.

Answer 1

Ok so it turns out that the threads being there in WAIT state were supposed to be there, as by default the threadpool size for Camel consumers is 10 (and in fact, I had at most 10 * my number of routes) threads.

Now, configuring the thread pool size is something that can be done, but it's not as easy as it seems, as there are several different ways to do so.

What fixed this for me is to set the threadPoolSize parameter in the rabbitMQ URI:

from("rabbitmq:$rabbitMQVhost?threadPoolSize=5&connectionFactory=#customConnectionFactory&queue=${it.rabbitMQQueue}

By doing so, the number of threads in WAIT state after several messages are processed on all routes is, as expected, 5 * my number of routes, which is better in my case: I don't have any big concurrency requirement but I have a significant number of routes, and having 10 hanging threads, with their memory footprint, for each route was draining my memory pretty fast.

Leaving this here as it looks like there's not much documentation around this topic and I had to bang my head for days on it.