title: Camel Split and Aggregate failing because messages going to multiple concurrent consumers
body: <p>I have a simple camel route that takes a list of items, splits them sending each element to a mq node for processing then joins them back together via an aggregator. </p>
  
  <p>Very close to the Composed Message Processor: <a href="http://camel.apache.org/composed-message-processor.html" rel="nofollow">http://camel.apache.org/composed-message-processor.html</a></p>
  
  <p>But we noticed that after the split, camel will create multiple concurrent consumers? or exchanges? Since the message is being sent to multiple consumers they never complete. </p>
  
  <p>List: 1,2,3,4</p>
  
  <p>Split: amq::process_each_item</p>
  
  <p>Aggregate:</p>
  
  <pre><code>[Camel (camel-3) thread #41 - Aggregating 1 - Waiting on 3 more items
  [Camel (camel-1) thread #16 - Aggregating 2 - Waiting on 3 more items
  [Camel (camel-3) thread #49 - Aggregating 3 - Waiting on 2 more items
  [Camel (camel-1) thread #15 - Aggregating 4 - Waiting on 2 more items
  </code></pre>
  
  <p>So, camel spawned 2 aggregators, and each is waiting on 4 items, but they only ever get two each.</p>
  
  <p>Camel Route: </p>
  
  <pre><code>&lt;camelContext xmlns="http://camel.apache.org/schema/spring"&gt;
  
          &lt;route&gt; &lt;!-- This route splits the reg request into it's items. Adding needed info to the message header.  --&gt;
              &lt;from uri="activemq:registration.splitByItemQueue" /&gt;  &lt;!-- pick up the reg req --&gt;
              &lt;setHeader headerName="regReqId"&gt; &lt;!-- Need to store the Reg Req in the header  --&gt;
                  &lt;simple&gt;${body.registrationRequest.id}&lt;/simple&gt;
              &lt;/setHeader&gt;
  
              &lt;split parallelProcessing="false" strategyRef="groupedExchangeAggregator"&gt; &lt;!-- Split the RegRequestInfo into it's individual requestItems (add, drop, etc) --&gt;
                  &lt;method ref="requestSplitter"  method="split" /&gt;   &lt;!-- does the actual splitting --&gt;
                  &lt;setHeader headerName="JMSXGroupID"&gt; &lt;!-- This is CRITICAL. It is how we ensure valid seat check counts without db locking --&gt;
                      &lt;simple&gt;FOID=${body.formatOfferingId}&lt;/simple&gt;  &lt;!-- grouping on the foid --&gt;
                  &lt;/setHeader&gt;
                  &lt;to uri="activemq:registration.lprActionQueue"/&gt; &lt;!-- send to queue's for processing--&gt;
              &lt;/split&gt;
          &lt;/route&gt;
  
          &lt;route&gt;    &lt;!-- performs the registration + seat check --&gt;
              &lt;from uri="activemq:registration.lprActionQueue" /&gt;
  
              &lt;bean ref="actionProcessor" method="process"/&gt; &lt;!-- go to the java code that makes all the decisions --&gt;
              &lt;to uri="activemq:registration.regReqItemJoinQueue"/&gt; &lt;!-- send to join queue's for final processing--&gt;
          &lt;/route&gt;
  
          &lt;route&gt;    &lt;!-- This route joins items from the reg req item split. Once all items have completed, update state--&gt;
              &lt;from uri="activemq:registration.regReqItemJoinQueue" /&gt;  &lt;!-- Every Reg Req Item will come here--&gt;
              &lt;aggregate strategyRef="groupedExchangeAggregator" ignoreInvalidCorrelationKeys="false" completionFromBatchConsumer="true"&gt; &lt;!-- take all the Reg Req Items an join them to their req --&gt;
                  &lt;correlationExpression&gt;
                      &lt;header&gt;regReqId&lt;/header&gt; &lt;!-- correlate on the regReqId we stored in the header --&gt;
                  &lt;/correlationExpression&gt;
  
                  &lt;bean ref="actionProcessor" method="updateRegistrationRequestStatus"/&gt; &lt;!-- update status --&gt;                   
              &lt;/aggregate&gt;
          &lt;/route&gt;
  &lt;/camelContext&gt;
  
  &lt;bean id="groupedExchangeAggregator" class="org.apache.camel.processor.aggregate.GroupedExchangeAggregationStrategy" /&gt;
  </code></pre>
  
  <p>On my local machine the above works fine, but when we deploy to our test server half of the messages go to one camel aggregator, half to the other. causing none to ever finish. Notice in the config below that we've set concurrent consumers to 1 for camel. </p>
  
  <p>Here's the camel / activemq config</p>
  
  <pre><code>&lt;amq:broker useJmx="false" persistent="false"&gt;
          &lt;amq:plugins&gt;
              &lt;amq:statisticsBrokerPlugin /&gt;
          &lt;/amq:plugins&gt;
          &lt;amq:transportConnectors&gt;
              &lt;amq:transportConnector uri="tcp://localhost:0" /&gt;
          &lt;/amq:transportConnectors&gt;
      &lt;/amq:broker&gt;
  
      &lt;!-- Basic AMQ connection factory --&gt;
      &lt;amq:connectionFactory id="amqConnectionFactory" brokerURL="vm://localhost" /&gt;
  
      &lt;!-- Wraps the AMQ connection factory in Spring's caching (ie: pooled) factory
           From the AMQ "Spring Support"-page: "You can use the PooledConnectionFactory for efficient pooling... or you
           can use the Spring JMS CachingConnectionFactory to achieve the same effect."
           See "Consuming JMS from inside Spring" at http://activemq.apache.org/spring-support.html
           Also see http://codedependents.com/2010/07/14/connectionfactories-and-caching-with-spring-and-activemq/
  
           Note: there are pros/cons to using Spring's caching factory vs Apache's PooledConnectionFactory; but, until
           we have more explicit reasons to favor one over the other, Spring's is less tightly-coupled to a specific
           AMQP-implementation.
           See http://stackoverflow.com/a/19594974
      --&gt;
      &lt;bean id="connectionFactory" class="org.springframework.jms.connection.CachingConnectionFactory"&gt;
          &lt;constructor-arg ref="amqConnectionFactory"/&gt;
          &lt;property name="sessionCacheSize" value="1"/&gt;
      &lt;/bean&gt;
  
      &lt;bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"&gt;
          &lt;constructor-arg ref="connectionFactory" /&gt;
      &lt;/bean&gt;
  
      &lt;bean id="jmsConfig"
            class="org.apache.camel.component.jms.JmsConfiguration"&gt;
          &lt;property name="connectionFactory" ref="connectionFactory"/&gt;
          &lt;property name="concurrentConsumers" value="1"/&gt;
      &lt;/bean&gt;
  
      &lt;bean id="activemq"
            class="org.apache.activemq.camel.component.ActiveMQComponent"&gt;
          &lt;property name="configuration" ref="jmsConfig"/&gt;
      &lt;/bean&gt;
  </code></pre>