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
