samza-dev mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Yan Fang <yanfang...@gmail.com>
Subject Re: Memory issue
Date Fri, 05 Jun 2015 23:42:41 GMT
Hi Davide,

Consumer takes the memory is because it maintains a cache for the messages, the size by default
is 50000. In your case, it maybe as big as 50000*12kb. See systems.system-name.
samza.fetch.threshold in http://samza.apache.org/learn/documentation/0.9/jobs/configuration-table.html

"if I slow down the task (for example by skipping CPU-intensive computation) "

When you skip some computation, shouldn't the task be faster ? If that's the case, I suspect
the task processes so fast that the consumer never reaches the 50000 threshold. But I haven't
verified it.

Thanks,

Yan Fang

> On Jun 5, 2015, at 3:50 AM, Davide Simoncelli <netcelli.tux@gmail.com> wrote:
> 
> Hello,
> 
> I have memory issues when running a task on the cluster. The Samza application performs
pattern matching on received messages and sends results to another Kafka topic.
> I tried to put some pressure on the system by sending messages with an average size of
12KiB. After a while containers crash because of no heap memory left or java.lang.OutOfMemoryError:
GC overhead limit exceeded.
> For example:
> 
> java.lang.OutOfMemoryError: Java heap space
>    at java.nio.HeapByteBuffer.<init>(HeapByteBuffer.java:57) ~[na:1.8.0_45]
>    at java.nio.ByteBuffer.allocate(ByteBuffer.java:335) ~[na:1.8.0_45]
>    at kafka.network.BoundedByteBufferReceive.byteBufferAllocate(BoundedByteBufferReceive.scala:80)
~[kafka_2.10-0.8.2.0.jar:na]
>    at kafka.network.BoundedByteBufferReceive.readFrom(BoundedByteBufferReceive.scala:63)
~[kafka_2.10-0.8.2.0.jar:na]
>    at kafka.network.Receive$class.readCompletely(Transmission.scala:56) ~[kafka_2.10-0.8.2.0.jar:na]
>    at kafka.network.BoundedByteBufferReceive.readCompletely(BoundedByteBufferReceive.scala:29)
~[kafka_2.10-0.8.2.0.jar:na]
>    at kafka.network.BlockingChannel.receive(BlockingChannel.scala:111) ~[kafka_2.10-0.8.2.0.jar:na]
>    at kafka.consumer.SimpleConsumer.liftedTree1$1(SimpleConsumer.scala:79) ~[kafka_2.10-0.8.2.0.jar:na]
>    at kafka.consumer.SimpleConsumer.kafka$consumer$SimpleConsumer$$sendRequest(SimpleConsumer.scala:68)
~[kafka_2.10-0.8.2.0.jar:na]
>    at kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply$mcV$sp(SimpleConsumer.scala:112)
~[kafka_2.10-0.8.2.0.jar:na]
>    at kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply(SimpleConsumer.scala:112)
~[kafka_2.10-0.8.2.0.jar:na]
>    at kafka.consumer.SimpleConsumer$$anonfun$fetch$1$$anonfun$apply$mcV$sp$1.apply(SimpleConsumer.scala:112)
~[kafka_2.10-0.8.2.0.jar:na]
>    at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33) ~[kafka_2.10-0.8.2.0.jar:na]
>    at kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply$mcV$sp(SimpleConsumer.scala:111)
~[kafka_2.10-0.8.2.0.jar:na]
>    at kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleConsumer.scala:111)
~[kafka_2.10-0.8.2.0.jar:na]
>    at kafka.consumer.SimpleConsumer$$anonfun$fetch$1.apply(SimpleConsumer.scala:111)
~[kafka_2.10-0.8.2.0.jar:na]
>    at kafka.metrics.KafkaTimer.time(KafkaTimer.scala:33) ~[kafka_2.10-0.8.2.0.jar:na]
>    at kafka.consumer.SimpleConsumer.fetch(SimpleConsumer.scala:110) ~[kafka_2.10-0.8.2.0.jar:na]
>    at org.apache.samza.system.kafka.DefaultFetchSimpleConsumer.fetch(DefaultFetchSimpleConsumer.scala:48)
~[samza-kafka_2.10-0.9.0.jar:na]
>    at org.apache.samza.system.kafka.DefaultFetchSimpleConsumer.defaultFetch(DefaultFetchSimpleConsumer.scala:41)
~[samza-kafka_2.10-0.9.0.jar:na]
>    at org.apache.samza.system.kafka.BrokerProxy.org$apache$samza$system$kafka$BrokerProxy$$fetchMessages(BrokerProxy.scala:176)
~[samza-kafka_2.10-0.9.0.jar:na]
>    at org.apache.samza.system.kafka.BrokerProxy$$anon$1$$anonfun$run$1.apply(BrokerProxy.scala:146)
~[samza-kafka_2.10-0.9.0.jar:na]
>    at org.apache.samza.system.kafka.BrokerProxy$$anon$1$$anonfun$run$1.apply(BrokerProxy.scala:133)
~[samza-kafka_2.10-0.9.0.jar:na]
>    at org.apache.samza.util.ExponentialSleepStrategy.run(ExponentialSleepStrategy.scala:82)
~[samza-core_2.10-0.9.0.jar:0.9.0]
>    at org.apache.samza.system.kafka.BrokerProxy$$anon$1.run(BrokerProxy.scala:132) ~[samza-kafka_2.10-0.9.0.jar:na]
>    at java.lang.Thread.run(Thread.java:745) ~[na:1.8.0_45]
> 
> java.lang.OutOfMemoryError: Java heap space
>    at kafka.utils.Utils$.readBytes(Utils.scala:127) ~[kafka_2.10-0.8.2.0.jar:na]
>    at kafka.utils.Utils$.readBytes(Utils.scala:121) ~[kafka_2.10-0.8.2.0.jar:na]
> 
> 
> So I instructed each container to generate a heap dump on crash. By inspecting them I
see the system consumer used a lot of memory:
> 
> http://imgur.com/W7i0dn6 <http://imgur.com/W7i0dn6>
> http://imgur.com/2CTphAk <http://imgur.com/2CTphAk>
> 
> Why does the system consumer use so much memory? I noticed if I slow down the task (for
example by skipping CPU-intensive computation) consumers don’t crash. Is that because there
is much more time so that the garbage collector can reclaim unused memory?
> 
> Regards
> 
> Davide

Mime
View raw message