camel-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Peter Keller (JIRA)" <>
Subject [jira] [Created] (CAMEL-10272) Aggregation is broken due to race condition in ParallelAggregateTask.doAggregateInternal()
Date Fri, 26 Aug 2016 07:47:21 GMT
Peter Keller created CAMEL-10272:

             Summary: Aggregation is broken due to race condition in ParallelAggregateTask.doAggregateInternal()
                 Key: CAMEL-10272
             Project: Camel
          Issue Type: Bug
          Components: camel-core
    Affects Versions: 2.17.3, 2.16.3
            Reporter: Peter Keller
            Priority: Critical

Unfortunately, I am not able to provide a (simple) unit test for comprehending the problem.
Furthermore our (complex) unit tests are not determinist due to the root cause of the problem.

However I tried to analyze the Camel Java code, to work out the problem. Please find below
my findings.

h3. Problem

The {{oldExchange}} is {{null}} more than once in the aggregator if a recipient list is processed
in parallel.

h3. Camel route

In my Camel route, a recipient list is worked of in parallel:
    .recipientList().method(new MyRecipientListBuilder())
        .aggregationStrategy(new MyAggregationStrategy())
    .bean(new MyPostProcessor());

Snippet of {{MyAggregationStrategy}}:
public Exchange aggregate(final Exchange oldExchange, final Exchange newExchange) {
    if (oldExchange == null) {
        // this is the case more than once which is not expected!
    // ...

{{oldExchange}} is null more than once which is not expected and which contradicts the contract
with Camel.

h3. Analysis

During the processing, Camel invokes {{MulticastProcessor.process()}}. Here the result object
{{AtomicExchange}} is created which is shared during the whole processing.

If the processing should be done in parallel (as it is the case for our route) then {{MulticastProcessor.doProcessParallel()}}
is invoked. Here one instance of {{AggregateOnTheFlyTask}} is initialized and {{aggregateOnTheFly()}}
is invoked *asynchronously* via {{run()}}  for *every* target in the recipient list.

In {{aggregateOnTheFly()}}, a new instance of {{ParallelAggregateTask}} is generated, and
if aggregation is not done in parallel (as it is the case in our route), {{}},
{{ParallelAggregateTask.doAggregate()}} (this method is synchronized), and 
{{ParallelAggregateTask.doAggregateInternal()}} is invoked:

protected void doAggregateInternal(AggregationStrategy strategy, AtomicExchange result, Exchange
exchange) {
    if (strategy != null) {
        // prepare the exchanges for aggregation
        Exchange oldExchange = result.get();
        ExchangeHelper.prepareAggregation(oldExchange, exchange);
        result.set(strategy.aggregate(oldExchange, exchange));

However, in {{ParallelAggregateTask.doAggregateInternal()}} there may occur a race condition
as {{result}} is shared by every instance of {{AggregateOnTheFlyTask}} such that {{oldExchange
= result.get()}} 
may be {{null}} more than once!

Note: As a new instance of {{ParallelAggregateTask}} for every target in recipient list is
created, the {{synchronized}} method {{ParallelAggregateTask.doAggregate()}} does not prevent
the race condition!

Does this sounds reasonably?

This message was sent by Atlassian JIRA

View raw message