[
https://issues.apache.org/activemq/browse/CAMEL-3337?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=63340#action_63340
]
Bryan Keller commented on CAMEL-3337:
-------------------------------------
I noticed in the code about setting pool size to 0 will create a synchronous
queue, however there is a check to ensure it is set to something greater than
zero and thus fails. Also, I tried creating a synchronous queue in my own
executor and it didn't work.
Finally, a persistent store will not solve the problem with data loss. The
persistent store is used for the aggregation cache, NOT the job queue. If you
have 500 completion jobs queued with the executor, they will be lost on server
shutdown even with persistence.
> 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
> Assignee: Claus Ibsen
> Priority: Minor
> Fix For: 2.6.0
>
>
> 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.
