Return-Path: 
 <dev-return-23264-apmail-camel-dev-archive=camel.apache.org@camel.apache.org>
Delivered-To: apmail-camel-dev-archive@www.apache.org
Received: (qmail 74849 invoked from network); 14 Nov 2010 07:49:15 -0000
Received: from unknown (HELO mail.apache.org) (140.211.11.3)
  by 140.211.11.9 with SMTP; 14 Nov 2010 07:49:15 -0000
Received: (qmail 23236 invoked by uid 500); 14 Nov 2010 07:49:46 -0000
Delivered-To: apmail-camel-dev-archive@camel.apache.org
Received: (qmail 23087 invoked by uid 500); 14 Nov 2010 07:49:46 -0000
Mailing-List: contact dev-help@camel.apache.org; run by ezmlm
Precedence: bulk
List-Help: <mailto:dev-help@camel.apache.org>
List-Unsubscribe: <mailto:dev-unsubscribe@camel.apache.org>
List-Post: <mailto:dev@camel.apache.org>
List-Id: <dev.camel.apache.org>
Reply-To: dev@camel.apache.org
Delivered-To: mailing list dev@camel.apache.org
Received: (qmail 23079 invoked by uid 500); 14 Nov 2010 07:49:45 -0000
Delivered-To: apmail-activemq-camel-dev@activemq.apache.org
Received: (qmail 23076 invoked by uid 99); 14 Nov 2010 07:49:45 -0000
Received: from athena.apache.org (HELO athena.apache.org) (140.211.11.136)
    by apache.org (qpsmtpd/0.29) with ESMTP; Sun, 14 Nov 2010 07:49:45 +0000
X-ASF-Spam-Status: No, hits=-2000.0 required=10.0
	tests=ALL_TRUSTED
X-Spam-Check-By: apache.org
Received: from [140.211.11.22] (HELO thor.apache.org) (140.211.11.22)
    by apache.org (qpsmtpd/0.29) with ESMTP; Sun, 14 Nov 2010 07:49:45 +0000
Received: from thor (localhost [127.0.0.1])
	by thor.apache.org (8.13.8+Sun/8.13.8) with ESMTP id oAE7nOWq009379
	for <camel-dev@activemq.apache.org>; Sun, 14 Nov 2010 07:49:24 GMT
Message-ID: <8113003.4731289720964366.JavaMail.jira@thor>
Date: Sun, 14 Nov 2010 02:49:24 -0500 (EST)
From: "Claus Ibsen (JIRA)" <jira@apache.org>
To: camel-dev@activemq.apache.org
Subject: [jira] Commented: (CAMEL-3337) Aggregator using an unbounded queue,
 can use up all heap
In-Reply-To: <18680561.4621289677285108.JavaMail.jira@thor>
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 7bit
X-JIRA-FingerPrint: ae95407df07c98740808b2ef9da0087c


    [ https://issues.apache.org/activemq/browse/CAMEL-3337?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=63338#action_63338 ] 

Claus Ibsen commented on CAMEL-3337:
------------------------------------

Yeah you should be able to set the parallel option and provide it a custom executor service which has a task queue with a limited size
http://camel.apache.org/threading-model.html



> Aggregator using an unbounded queue, can use up all heap
> --------------------------------------------------------
>
>                 Key: CAMEL-3337
>                 URL: https://issues.apache.org/activemq/browse/CAMEL-3337
>             Project: Apache Camel
>          Issue Type: Improvement
>          Components: camel-core
>    Affects Versions: 2.5.0
>         Environment: JDK 1.6.0_22 (Windows 7 64-bit and OS X 10.6.5)
>            Reporter: Bryan Keller
>            Priority: Minor
>
> My app is having memory issues due to the use of an aggregator. My app is fairly straightforward. It reads from a CSV using streaming, tokenizes it by line, passes the result to a processor, aggregates the result, then puts this on a JMS queue. Here is the route definition (for Spring):
> {code:xml}
> <route>
>   <from "file:in" />
>   <split streaming="true">
>     <tokenize token="\n" regex="false" />
>     <unmarshal><csv /></unmarshal>
>     <bean ref="myBean" method="translate" />
>     <aggregate strategyRef="myAggregationStrategy">
>     <correlationExpression><constant>true</constant></correlationExpression>
>     <completionTimeout><simple>1000</simple></completionTimeout>
>     <completionSize><simple>100</simple></completionSize>
>     <to uri="activemq:queue:myQueue"/>   		
>   </aggregate>
>   </split>
> </route>
> {code}
> The problem happens when the consumer of "myQueue" is not as fast as the file reading and parsing. With a slow consumer, ActiveMQ will eventually throttle the producer so the message queue doesn't use up all memory and/or disk space.
> As messages are passed to the aggregator, it internally submits jobs to an executor which will then put the message in the queue. This executor uses an unbounded queue. If the message producer has been throttled, then the process jobs cannot queue the messages quickly enough, and the executor's queue will continue to back up until all memory is used.
> As a workaround, I am thinking I could implement my own executor service which is synchronous or at least blocks when the queue size reaches a certain size. I haven't yet firgured out how to configure this however.

-- 
This message is automatically generated by JIRA.
-
You can reply to this email to add a comment to the issue online.