You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@camel.apache.org by ebinsingh <eb...@VerizonWireless.com> on 2012/04/27 17:25:35 UTC
Java heap space issue with Aggregation
Hi All,
I am trying to aggregate large number of xml files into files of 50000
records.
I am getting java.lang.OutOfMemoryError - Java heap space error.
I am trying to see if there are any leaks but to my eyes i do not see any.
Appreciate your thoughts on this.
Aggreation logic:
public class GlobalAggrStratergy implements AggregationStrategy {
private static Logger log = Logger.getLogger(GlobalAggrStratergy.class);
int counter = 0;
@Override
public Exchange aggregate(Exchange exchange1, Exchange exchange2) {
try{
StringBuilder builder;
if (exchange1 == null || null == exchange1.getIn().getBody()) {
builder = new StringBuilder();
exchange1 = new DefaultExchange(new DefaultCamelContext());
exchange1.getIn().setBody(builder);
}
builder = exchange1.getIn().getBody(StringBuilder.class);
builder.append(exchange2.getIn().getBody()+"\n");
exchange1.getIn().setBody(builder);
exchange1.getIn().setHeader(Exchange.FILE_NAME_ONLY,
exchange2.getProperty(Exchange.FILE_NAME_ONLY));
counter++;
}catch(Exception ex){
log.error("Error aggregating", ex);
}
exchange1.setProperty(Exchange.BATCH_SIZE, counter);
if(counter >= 50000)
counter = 0;
return exchange1;
}
Route configuration:
public void configure() throws Exception
{
from("direct:producerQueue").log("File name: ${in.header.fileName}")
.setProperty(Exchange.FILE_NAME_ONLY, simple("${file:onlyname.noext}"))
.split().tokenizeXML("IPDR").streaming()
.aggregate(header("messageId"), new
GlobalAggrStratergy()).completionSize(50000).completionTimeout(20000)
.process(new IPDRHeaderFooterProcessor())
.to(IPDRUtil.getInstance().getProperty("IPDROutputDir"));
}
Thanks & regards,
Ebe
--
View this message in context: http://camel.465427.n5.nabble.com/Java-heap-space-issue-with-Aggregation-tp5670608p5670608.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Java heap space issue with Aggregation
Posted by Christian Müller <ch...@gmail.com>.
How large is the payload of one exchange?
Can you increase your memory, e.G. -Xms2048m -Xmx4094m?
When I remember right, 4GB is the maximum for 32 bit OS...
Best,
Christian
On Fri, Apr 27, 2012 at 10:09 PM, ebinsingh <
ebenezer.singh@verizonwireless.com> wrote:
> I have set it as
>
> -Xms1024m -Xmx1024m
>
> --
> View this message in context:
> http://camel.465427.n5.nabble.com/Java-heap-space-issue-with-Aggregation-tp5670608p5671375.html
> Sent from the Camel - Users mailing list archive at Nabble.com.
>
Re: Java heap space issue with Aggregation
Posted by ebinsingh <eb...@VerizonWireless.com>.
I have set it as
-Xms1024m -Xmx1024m
--
View this message in context: http://camel.465427.n5.nabble.com/Java-heap-space-issue-with-Aggregation-tp5670608p5671375.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Java heap space issue with Aggregation
Posted by Christian Müller <ch...@gmail.com>.
What are your JVM settings regarding to the memory?
How large is the payload of one exchange?
Best,
Christian
On Fri, Apr 27, 2012 at 7:33 PM, ebinsingh <
ebenezer.singh@verizonwireless.com> wrote:
> After running Eclipse Memory analyser, I see the below in the analysis data
> ...
>
> One instance of "char[]" loaded by "<system class loader>" occupies
> 63,586,312 (87.23%) bytes
>
>
> Please can someone help me is fixing this.
>
> Thanks & regards,
> Ebe
>
> --
> View this message in context:
> http://camel.465427.n5.nabble.com/Java-heap-space-issue-with-Aggregation-tp5670608p5670938.html
> Sent from the Camel - Users mailing list archive at Nabble.com.
>
Re: Java heap space issue with Aggregation
Posted by ebinsingh <eb...@VerizonWireless.com>.
After running Eclipse Memory analyser, I see the below in the analysis data
...
One instance of "char[]" loaded by "<system class loader>" occupies
63,586,312 (87.23%) bytes
Please can someone help me is fixing this.
Thanks & regards,
Ebe
--
View this message in context: http://camel.465427.n5.nabble.com/Java-heap-space-issue-with-Aggregation-tp5670608p5670938.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Java heap space issue with Aggregation
Posted by ebinsingh <eb...@VerizonWireless.com>.
This is the error trace I see in my logs ...
org.apache.camel.CamelExchangeException: Error occurred during aggregation.
Exchange[null]. Caused by: [java.lang.OutOfMemoryError - Java heap space]
at
org.apache.camel.processor.aggregate.AggregateProcessor.doAggregation(AggregateProcessor.java:243)
at
org.apache.camel.processor.aggregate.AggregateProcessor.process(AggregateProcessor.java:197)
at
org.apache.camel.util.AsyncProcessorConverterHelper$ProcessorToAsyncProcessorBridge.process(AsyncProcessorConverterHelper.java:61)
at
org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:73)
at
org.apache.camel.processor.DelegateAsyncProcessor.processNext(DelegateAsyncProcessor.java:99)
at
org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:90)
at
org.apache.camel.management.InstrumentationProcessor.process(InstrumentationProcessor.java:71)
at
org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:73)
at
org.apache.camel.processor.DelegateAsyncProcessor.processNext(DelegateAsyncProcessor.java:99)
at
org.apache.camel.processor.DelegateAsyncProcessor.process(DelegateAsyncProcessor.java:90)
at
org.apache.camel.processor.interceptor.TraceInterceptor.process(TraceInterceptor.java:91)
at
org.apache.camel.util.AsyncProcessorHelper.process(AsyncProcessorHelper.java:73)
at
org.apache.camel.processor.RedeliveryErrorHandler.processErrorHandler(RedeliveryErrorHandler.java:
--
View this message in context: http://camel.465427.n5.nabble.com/Java-heap-space-issue-with-Aggregation-tp5670608p5670850.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Java heap space issue with Aggregation
Posted by Claus Ibsen <cl...@gmail.com>.
Hi
Whatever you do with the aggregator is that it holds the aggregated
data in memory, so you need to be able to store that in memory.
You could use a persistent repository, and use the claim check EIP
pattern and only store "tokens" to the actual data which is stored
offline (eg not in memory) for very large data.
That said, yesterday I worked on adding a group option to the splitter
+ tokenizer, so you can instruct it to split the data by N number of
tokens (eg lines, XML pairs etc.).
https://issues.apache.org/jira/browse/CAMEL-5236
I got the regular stuff implemented, but haven't got around the
tokenizerXML yet. As well I want to add an option so you can tell it
to include the parent tag as well.
eg
<orders>
<order>
...
</order>
...
a zillion of <order>
</orders>
And then you can group that by 1000, so you get 1000 <order> in a <orders> tag
On Fri, Apr 27, 2012 at 5:25 PM, ebinsingh
<eb...@verizonwireless.com> wrote:
> Hi All,
>
> I am trying to aggregate large number of xml files into files of 50000
> records.
> I am getting java.lang.OutOfMemoryError - Java heap space error.
>
> I am trying to see if there are any leaks but to my eyes i do not see any.
>
> Appreciate your thoughts on this.
>
> Aggreation logic:
>
> public class GlobalAggrStratergy implements AggregationStrategy {
> private static Logger log = Logger.getLogger(GlobalAggrStratergy.class);
> int counter = 0;
> @Override
> public Exchange aggregate(Exchange exchange1, Exchange exchange2) {
> try{
> StringBuilder builder;
> if (exchange1 == null || null == exchange1.getIn().getBody()) {
> builder = new StringBuilder();
> exchange1 = new DefaultExchange(new DefaultCamelContext());
> exchange1.getIn().setBody(builder);
> }
> builder = exchange1.getIn().getBody(StringBuilder.class);
> builder.append(exchange2.getIn().getBody()+"\n");
> exchange1.getIn().setBody(builder);
> exchange1.getIn().setHeader(Exchange.FILE_NAME_ONLY,
> exchange2.getProperty(Exchange.FILE_NAME_ONLY));
> counter++;
> }catch(Exception ex){
> log.error("Error aggregating", ex);
> }
> exchange1.setProperty(Exchange.BATCH_SIZE, counter);
> if(counter >= 50000)
> counter = 0;
> return exchange1;
> }
>
>
> Route configuration:
>
> public void configure() throws Exception
> {
> from("direct:producerQueue").log("File name: ${in.header.fileName}")
> .setProperty(Exchange.FILE_NAME_ONLY, simple("${file:onlyname.noext}"))
> .split().tokenizeXML("IPDR").streaming()
> .aggregate(header("messageId"), new
> GlobalAggrStratergy()).completionSize(50000).completionTimeout(20000)
> .process(new IPDRHeaderFooterProcessor())
> .to(IPDRUtil.getInstance().getProperty("IPDROutputDir"));
> }
>
> Thanks & regards,
> Ebe
>
> --
> View this message in context: http://camel.465427.n5.nabble.com/Java-heap-space-issue-with-Aggregation-tp5670608p5670608.html
> Sent from the Camel - Users mailing list archive at Nabble.com.
--
Claus Ibsen
-----------------
CamelOne 2012 Conference, May 15-16, 2012: http://camelone.com
FuseSource
Email: cibsen@fusesource.com
Web: http://fusesource.com
Twitter: davsclaus, fusenews
Blog: http://davsclaus.blogspot.com/
Author of Camel in Action: http://www.manning.com/ibsen/
Re: Java heap space issue with Aggregation
Posted by Christian Müller <ch...@gmail.com>.
What's about this:
public class GlobalAggrStratergy implements AggregationStrategy {
private static final transient Logger log =
LoggerFactory.getLogger(GlobalAggrStratergy.class);
private int counter = 0;
@Override
public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
try {
log.info("oldExchange: {}", oldExchange);
log.info("newExchange: {}", newExchange);
// happens only the first time
if (oldExchange == null) {
// create and initialize the StringBuilder
newExchange.getIn().setBody(new
StringBuilder(newExchange.getIn().getBody(String.class)));
return newExchange;
}
// append the new payload to the old exchange
StringBuilder stringBuilder =
oldExchange.getIn().getBody(StringBuilder.class);
stringBuilder.append("\n");
stringBuilder.append(newExchange.getIn().getBody(String.class));
counter++;
} catch (Exception ex) {
throw new RuntimeException(ex);
}
oldExchange.setProperty("AGGREGATION_COUNTER", counter);
if (counter == 50000) {
counter = 0;
}
return oldExchange;
}
}
Best,
Christian
On Mon, Apr 30, 2012 at 4:50 PM, ebinsingh <
ebenezer.singh@verizonwireless.com> wrote:
> I am very sorry for not understanding this properly.
> Please accept my apologies if am wasting your time.
>
> I am totally missing something here.
> Below is the change I had made (Hopeing that this is what you meant by
> saying Mutate the old exchange).
>
> I am using StringBuilder so that the process os faster and also helps not
> creating many String objects.
>
> 1. when I use String.class all works fine until am running out of
> heapspace.
> 2. If I use StringBuilder, I get the following nullPointer.
> 3. Looks like there is no converter type to conver from String to
> StringBuilder.
>
>
> 2012-04-30 10:34:38,087 INFO [Camel (camel-1) thread #2 -
> seda://streamQueue] ipdr.GlobalAggrStratergy - is oldExchange null:
> Exchange[null] AND is newExchange null: Exchange[null]
> 2012-04-30 10:34:38,087 INFO [Camel (camel-1) thread #2 -
> seda://streamQueue] ipdr.GlobalAggrStratergy - Aggregate old orders: null
> 2012-04-30 10:34:38,087 INFO [Camel (camel-1) thread #2 -
> seda://streamQueue] ipdr.GlobalAggrStratergy - Aggregate new order: null
> 2012-04-30 10:34:38,087 ERROR [Camel (camel-1) thread #2 -
> seda://streamQueue] ipdr.GlobalAggrStratergy - Error aggregating
> java.lang.NullPointerException
> at
> com.vzw.fp.ipdr.GlobalAggrStratergy.aggregate(GlobalAggrStratergy.java:26)
>
>
>
> @Override
> public Exchange aggregate(Exchange oldExchange, Exchange
> newExchange) {
> try{
> log.info(" is oldExchange null: "+oldExchange+"
> AND is newExchange null:
> "+newExchange);
> if (oldExchange == null) {
> return newExchange;
> } else {
> log.info("Aggregate old orders: " +
> oldExchange.getIn().getBody(StringBuilder.class));
> log.info("Aggregate new order: " +
> newExchange.getIn().getBody(StringBuilder.class));
> oldExchange.getIn().setBody(
>
> oldExchange.getIn().getBody(StringBuilder.class).
>
> append(newExchange.getIn().getBody(StringBuilder.class)+"\n"));
> counter++;
> }
> }catch(Exception ex){
> log.error("Error aggregating", ex);
> }
> oldExchange.setProperty(Exchange.CONTENT_LENGTH, counter);
> if(counter >= 50000)
> counter = 0;
> oldExchange.getIn().setHeader(Exchange.FILE_NAME_ONLY,
> newExchange.getProperty(Exchange.FILE_NAME_ONLY));
> return oldExchange;
> }
>
> --
> View this message in context:
> http://camel.465427.n5.nabble.com/Java-heap-space-issue-with-Aggregation-tp5670608p5676110.html
> Sent from the Camel - Users mailing list archive at Nabble.com.
>
Re: Java heap space issue with Aggregation
Posted by ebinsingh <eb...@VerizonWireless.com>.
I am very sorry for not understanding this properly.
Please accept my apologies if am wasting your time.
I am totally missing something here.
Below is the change I had made (Hopeing that this is what you meant by
saying Mutate the old exchange).
I am using StringBuilder so that the process os faster and also helps not
creating many String objects.
1. when I use String.class all works fine until am running out of heapspace.
2. If I use StringBuilder, I get the following nullPointer.
3. Looks like there is no converter type to conver from String to
StringBuilder.
2012-04-30 10:34:38,087 INFO [Camel (camel-1) thread #2 -
seda://streamQueue] ipdr.GlobalAggrStratergy - is oldExchange null:
Exchange[null] AND is newExchange null: Exchange[null]
2012-04-30 10:34:38,087 INFO [Camel (camel-1) thread #2 -
seda://streamQueue] ipdr.GlobalAggrStratergy - Aggregate old orders: null
2012-04-30 10:34:38,087 INFO [Camel (camel-1) thread #2 -
seda://streamQueue] ipdr.GlobalAggrStratergy - Aggregate new order: null
2012-04-30 10:34:38,087 ERROR [Camel (camel-1) thread #2 -
seda://streamQueue] ipdr.GlobalAggrStratergy - Error aggregating
java.lang.NullPointerException
at
com.vzw.fp.ipdr.GlobalAggrStratergy.aggregate(GlobalAggrStratergy.java:26)
@Override
public Exchange aggregate(Exchange oldExchange, Exchange newExchange) {
try{
log.info(" is oldExchange null: "+oldExchange+" AND is newExchange null:
"+newExchange);
if (oldExchange == null) {
return newExchange;
} else {
log.info("Aggregate old orders: " +
oldExchange.getIn().getBody(StringBuilder.class));
log.info("Aggregate new order: " +
newExchange.getIn().getBody(StringBuilder.class));
oldExchange.getIn().setBody(
oldExchange.getIn().getBody(StringBuilder.class).
append(newExchange.getIn().getBody(StringBuilder.class)+"\n"));
counter++;
}
}catch(Exception ex){
log.error("Error aggregating", ex);
}
oldExchange.setProperty(Exchange.CONTENT_LENGTH, counter);
if(counter >= 50000)
counter = 0;
oldExchange.getIn().setHeader(Exchange.FILE_NAME_ONLY,
newExchange.getProperty(Exchange.FILE_NAME_ONLY));
return oldExchange;
}
--
View this message in context: http://camel.465427.n5.nabble.com/Java-heap-space-issue-with-Aggregation-tp5670608p5676110.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Java heap space issue with Aggregation
Posted by ebinsingh <eb...@VerizonWireless.com>.
Hi Claus,
I aplogise for not understanding your comments (mutate the existing
oldExchange). Please help me understand if you have sometime.
I am doing as I did because of the following ...
1. I guess that the old exchange is null only once for every aggregation
block (i.e Since I have the completion size as 50000, the old exchange will
be null once every 5000 data). I guess this would not be a overhead causing
the outage.
2. Since there is no defauls converter to StringBuilder, I had to make the
oldExchange contents as StringBuilder as it enhances the performance by a
big margin.
But now have come to a conclusion that Java cannot handle this big a data
(50000 payload). But the concern really is that it was not able to handle
5000 payloads. Size of a payload would be equal or less than a swaps product
entity.
Thanks & regards,
Ebe
--
View this message in context: http://camel.465427.n5.nabble.com/Java-heap-space-issue-with-Aggregation-tp5670608p5676295.html
Sent from the Camel - Users mailing list archive at Nabble.com.
Re: Java heap space issue with Aggregation
Posted by Claus Ibsen <cl...@gmail.com>.
On Sat, Apr 28, 2012 at 5:18 AM, Willem Jiang <wi...@gmail.com> wrote:
> The exchange property of the Exchange.BATCH_SIZE will be used when the
> aggregation call the isCompeleted(), So please don't change it dynamically.
>
Yes I told him to not create a new DefaultExchange all the time. But
only mutate the existing oldExchange.
And if he uses a StringBuilder he can append new data to it, instead
of creating a new instance all the time as well.
>
> On 4/27/12 11:25 PM, ebinsingh wrote:
>>
>> Hi All,
>>
>> I am trying to aggregate large number of xml files into files of 50000
>> records.
>> I am getting java.lang.OutOfMemoryError - Java heap space error.
>>
>> I am trying to see if there are any leaks but to my eyes i do not see any.
>>
>> Appreciate your thoughts on this.
>>
>> Aggreation logic:
>>
>> public class GlobalAggrStratergy implements AggregationStrategy {
>> private static Logger log =
>> Logger.getLogger(GlobalAggrStratergy.class);
>> int counter = 0;
>> @Override
>> public Exchange aggregate(Exchange exchange1, Exchange exchange2) {
>> try{
>> StringBuilder builder;
>> if (exchange1 == null || null ==
>> exchange1.getIn().getBody()) {
>> builder = new StringBuilder();
>> exchange1 = new DefaultExchange(new
>> DefaultCamelContext());
>> exchange1.getIn().setBody(builder);
>> }
>> builder = exchange1.getIn().getBody(StringBuilder.class);
>> builder.append(exchange2.getIn().getBody()+"\n");
>> exchange1.getIn().setBody(builder);
>> exchange1.getIn().setHeader(Exchange.FILE_NAME_ONLY,
>> exchange2.getProperty(Exchange.FILE_NAME_ONLY));
>> counter++;
>> }catch(Exception ex){
>> log.error("Error aggregating", ex);
>> }
>> exchange1.setProperty(Exchange.BATCH_SIZE, counter);
>> if(counter>= 50000)
>> counter = 0;
>> return exchange1;
>> }
>>
>>
>> Route configuration:
>>
>> public void configure() throws Exception
>> {
>> from("direct:producerQueue").log("File name:
>> ${in.header.fileName}")
>> .setProperty(Exchange.FILE_NAME_ONLY,
>> simple("${file:onlyname.noext}"))
>> .split().tokenizeXML("IPDR").streaming()
>> .aggregate(header("messageId"), new
>> GlobalAggrStratergy()).completionSize(50000).completionTimeout(20000)
>> .process(new IPDRHeaderFooterProcessor())
>> .to(IPDRUtil.getInstance().getProperty("IPDROutputDir"));
>> }
>>
>> Thanks& regards,
>>
>> Ebe
>>
>> --
>> View this message in context:
>> http://camel.465427.n5.nabble.com/Java-heap-space-issue-with-Aggregation-tp5670608p5670608.html
>> Sent from the Camel - Users mailing list archive at Nabble.com.
>>
>
>
> --
> Willem
> ----------------------------------
> CamelOne 2012 Conference, May 15-16, 2012: http://camelone.com
> FuseSource
> Web: http://www.fusesource.com
> Blog: http://willemjiang.blogspot.com (English)
> http://jnn.javaeye.com (Chinese)
> Twitter: willemjiang
> Weibo: willemjiang
--
Claus Ibsen
-----------------
CamelOne 2012 Conference, May 15-16, 2012: http://camelone.com
FuseSource
Email: cibsen@fusesource.com
Web: http://fusesource.com
Twitter: davsclaus, fusenews
Blog: http://davsclaus.blogspot.com/
Author of Camel in Action: http://www.manning.com/ibsen/
Re: Java heap space issue with Aggregation
Posted by Willem Jiang <wi...@gmail.com>.
The exchange property of the Exchange.BATCH_SIZE will be used when the
aggregation call the isCompeleted(), So please don't change it dynamically.
On 4/27/12 11:25 PM, ebinsingh wrote:
> Hi All,
>
> I am trying to aggregate large number of xml files into files of 50000
> records.
> I am getting java.lang.OutOfMemoryError - Java heap space error.
>
> I am trying to see if there are any leaks but to my eyes i do not see any.
>
> Appreciate your thoughts on this.
>
> Aggreation logic:
>
> public class GlobalAggrStratergy implements AggregationStrategy {
> private static Logger log = Logger.getLogger(GlobalAggrStratergy.class);
> int counter = 0;
> @Override
> public Exchange aggregate(Exchange exchange1, Exchange exchange2) {
> try{
> StringBuilder builder;
> if (exchange1 == null || null == exchange1.getIn().getBody()) {
> builder = new StringBuilder();
> exchange1 = new DefaultExchange(new DefaultCamelContext());
> exchange1.getIn().setBody(builder);
> }
> builder = exchange1.getIn().getBody(StringBuilder.class);
> builder.append(exchange2.getIn().getBody()+"\n");
> exchange1.getIn().setBody(builder);
> exchange1.getIn().setHeader(Exchange.FILE_NAME_ONLY,
> exchange2.getProperty(Exchange.FILE_NAME_ONLY));
> counter++;
> }catch(Exception ex){
> log.error("Error aggregating", ex);
> }
> exchange1.setProperty(Exchange.BATCH_SIZE, counter);
> if(counter>= 50000)
> counter = 0;
> return exchange1;
> }
>
>
> Route configuration:
>
> public void configure() throws Exception
> {
> from("direct:producerQueue").log("File name: ${in.header.fileName}")
> .setProperty(Exchange.FILE_NAME_ONLY, simple("${file:onlyname.noext}"))
> .split().tokenizeXML("IPDR").streaming()
> .aggregate(header("messageId"), new
> GlobalAggrStratergy()).completionSize(50000).completionTimeout(20000)
> .process(new IPDRHeaderFooterProcessor())
> .to(IPDRUtil.getInstance().getProperty("IPDROutputDir"));
> }
>
> Thanks& regards,
> Ebe
>
> --
> View this message in context: http://camel.465427.n5.nabble.com/Java-heap-space-issue-with-Aggregation-tp5670608p5670608.html
> Sent from the Camel - Users mailing list archive at Nabble.com.
>
--
Willem
----------------------------------
CamelOne 2012 Conference, May 15-16, 2012: http://camelone.com
FuseSource
Web: http://www.fusesource.com
Blog: http://willemjiang.blogspot.com (English)
http://jnn.javaeye.com (Chinese)
Twitter: willemjiang
Weibo: willemjiang