title: How to use Aggregator correctly in apache camel to insert huge data into mongodb
body: <p>I am using apache camel to read huge XMLTYPE data from oracle DB (about 500M rows), process it and insert it into mongodb. I am using splitter for processing and the output message is in <strong>org.bson.Document</strong>.
  The problem is with the aggregator, it returns either an exception  <strong>Cannot deserialize as out of START_ARRAY token</strong> or the number of inserted rows is very few (just 9 rows).</p>
  <p>This the aggregator's code:</p>
  <pre class="lang-java prettyprint-override"><code>public class dAggregator implements AggregationStrategy {
      @Override
      public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
          Document newBody = newExchange.getIn().getBody(Document.class);
          ArrayList&lt;Document&gt; list = new ArrayList&lt;&gt;();
          if (oldExchange == null) {
              list.add(newBody);
              newExchange.getIn().setBody(list);
              return newExchange;
          } else {
              list.add(oldExchange.getIn().getBody(Document.class));
              return oldExchange;
          }
      }
  }
  </code></pre>
  <p>Camel Context:</p>
  <pre class="lang-xml prettyprint-override"><code>&lt;camelContext xmlns=&quot;http://camel.apache.org/schema/spring&quot;&gt;
      
      &lt;threadPool id=&quot;myPool&quot; rejectedPolicy=&quot;CallerRuns&quot; threadName=&quot;spl&quot; maxQueueSize=&quot;120&quot; maxPoolSize=&quot;250&quot; poolSize=&quot;80&quot;/&gt;
      &lt;threadPool id=&quot;aggPool&quot; rejectedPolicy=&quot;CallerRuns&quot; threadName=&quot;agg&quot; maxQueueSize=&quot;120&quot; maxPoolSize=&quot;150&quot; poolSize=&quot;50&quot;/&gt;
       
      &lt;route streamCache=&quot;true&quot;&gt;
          &lt;from uri=&quot;timer://foo?repeatCount=1&quot;/&gt;
              &lt;setBody&gt;
                  &lt;constant&gt;SELECT * FROM CUSTOMER&lt;/constant&gt;
              &lt;/setBody&gt;
              &lt;to id=&quot;sql&quot; uri=&quot;jdbc:myds?outputType=StreamList&amp;resetAutoCommit=false&amp;statement.fetchSize=100000&quot;/&gt;
              &lt;split parallelProcessing=&quot;true&quot; executorServiceRef=&quot;myPool&quot; stopOnException=&quot;true&quot; streaming=&quot;true&quot;&gt;
                  &lt;simple&gt;${in.body}&lt;/simple&gt;
                  &lt;process ref=&quot;rowProcessor&quot;/&gt;
                  &lt;to uri=&quot;seda:toinsert?size=1000000&amp;amp;blockWhenFull=true&quot;/&gt;
              &lt;/split&gt;
      &lt;/route&gt;
  
      &lt;route streamCache=&quot;true&quot;&gt;
          &lt;from uri=&quot;seda:toinsert&quot;/&gt;
              &lt;aggregate completionSize=&quot;50000&quot; completionTimeout=&quot;30000&quot; id=&quot;aggregate_1&quot; strategyRef=&quot;aggStrategy&quot; parallelProcessing=&quot;true&quot; executorServiceRef=&quot;aggPool&quot;&gt;
              &lt;correlationExpression&gt;
                  &lt;constant&gt;true&lt;/constant&gt;
              &lt;/correlationExpression&gt;
              &lt;to uri=&quot;mongodb:mongoBean?database=mydb&amp;collection=customer&amp;operation=insert&quot;/&gt;
      &lt;/route&gt;
  
  &lt;/camelContext&gt;
  </code></pre>
  <p>Is there anything wrong or missing? Any thoughts or tips are more than welcome.</p>