Camel Aggregation wait for lock

Question

I have a Camel 2.25.0 route listening to a JMS queue which aggregates on a key. The aggregation is configured with a 10 minutes completion timeout, a LevelDBAggregationRepository and the UseLatestAggregationStrategy.

from("jms:inbound")
        .log("Received views change stream")
        .bean(viewService, "updateDependencies")
        .aggregate()
        .header("JMSXGroupID")
        .aggregationStrategy(new UseLatestAggregationStrategy())
        .completionTimeout(600000) // 10 minutes
        .completeAllOnStop()
        .aggregationRepository(new LevelDBAggregationRepository("aggregation-repository", "./aggregation-repository"))
        .log("Process views change stream")
        .to("jms:outbound");

From time to time, the thread is blocked and the queue is not consumed. When it happens the logs show that there is a wait between the updateDependencies call and the start of the aggregation. XXXX is the end of the previous aggregation, and YYYY is the blocked aggregation.

APM duration: 1,073.2s
August 7th 2020, 11:39:20.327   onAggregation +++  end  +++ with correlation key: XXXX
August 7th 2020, 11:39:20.327   Received views change stream
August 7th 2020, 11:39:20.341   1 documents touched based on update // log from updateDependencies
August 7th 2020, 11:57:13.555   onAggregation +++ start +++ with correlation key: YYYY

In AggregateProcessor, there is a lock before the aggregation

lock.lock();
try {
    aggregated = doAggregation(key, copy);
} finally {
    lock.unlock();
}

My guess is that the wait time is due to the lock but it does not make sense since we can see that the previous aggregation (XXXX) is finished, so the lock should be released.

I also checked the application, and other threads are still running.

Camel Aggregation wait for lock

0 Answers