Hi, Thanks for the logs, however I do not see before mentioned exceptions in it. It ends with java.lang.InterruptedException
Is it the correct log file? Also, could you attach the std output file of the failing TaskManager? Piotrek > On 10 Nov 2017, at 08:42, ÇETİNKAYA EBRU ÇETİNKAYA EBRU > <b20926...@cs.hacettepe.edu.tr> wrote: > > On 2017-11-09 20:08, Piotr Nowojski wrote: >> Hi, >> Could you attach full logs from those task managers? At first glance I >> don’t see a connection between those exceptions and any memory issue >> that you might had. It looks like a dependency issue in one (some? >> All?) of your jobs. >> Did you build your jars with -Pbuild-jar profile as described here: >> https://ci.apache.org/projects/flink/flink-docs-release-1.3/quickstart/java_api_quickstart.html#build-project >> ? >> If that doesn’t help. Can you binary search which job is causing the >> problem? There might be some Flink incompatibility between different >> versions and rebuilding a job’s jar with a version matching to the >> cluster version might help. >> Piotrek >>> On 9 Nov 2017, at 17:36, ÇETİNKAYA EBRU ÇETİNKAYA EBRU >>> <b20926...@cs.hacettepe.edu.tr> wrote: >>> On 2017-11-08 18:30, Piotr Nowojski wrote: >>> Btw, Ebru: >>> I don’t agree that the main suspect is NetworkBufferPool. On your >>> screenshots it’s memory consumption was reasonable and stable: >>> 596MB >>> -> 602MB -> 597MB. >>> PoolThreadCache memory usage ~120MB is also reasonable. >>> Do you experience any problems, like Out Of Memory >>> errors/crashes/long >>> GC pauses? Or just JVM process is using more memory over time? You >>> are >>> aware that JVM doesn’t like to release memory back to OS once it >>> was >>> used? So increasing memory usage until hitting some limit (for >>> example >>> JVM max heap size) is expected behaviour. >>> Piotrek >>> On 8 Nov 2017, at 15:48, Piotr Nowojski <pi...@data-artisans.com> >>> wrote: >>> I don’t know if this is relevant to this issue, but I was >>> constantly getting failures trying to reproduce this leak using your >>> Job, because you were using non deterministic getKey function: >>> @Override >>> public Integer getKey(Integer event) { >>> Random randomGen = new Random((new Date()).getTime()); >>> return randomGen.nextInt() % 8; >>> } >>> And quoting Java doc of KeySelector: >>> "If invoked multiple times on the same object, the returned key must >>> be the same.” >>> I’m trying to reproduce this issue with following job: >>> https://gist.github.com/pnowojski/b80f725c1af7668051c773438637e0d3 >>> Where IntegerSource is just an infinite source, DisardingSink is >>> well just discarding incoming data. I’m cancelling the job every 5 >>> seconds and so far (after ~15 minutes) my memory consumption is >>> stable, well below maximum java heap size. >>> Piotrek >>> On 8 Nov 2017, at 15:28, Javier Lopez <javier.lo...@zalando.de> >>> wrote: >>> Yes, I tested with just printing the stream. But it could take a >>> lot of time to fail. >>> On Wednesday, 8 November 2017, Piotr Nowojski >>> <pi...@data-artisans.com> wrote: >>> Thanks for quick answer. >>> So it will also fail after some time with `fromElements` source >>> instead of Kafka, right? >>> Did you try it also without a Kafka producer? >>> Piotrek >>> On 8 Nov 2017, at 14:57, Javier Lopez <javier.lo...@zalando.de> >>> wrote: >>> Hi, >>> You don't need data. With data it will die faster. I tested as >>> well with a small data set, using the fromElements source, but it >>> will take some time to die. It's better with some data. >>> On 8 November 2017 at 14:54, Piotr Nowojski >>> <pi...@data-artisans.com> wrote: >>> Hi, >>> Thanks for sharing this job. >>> Do I need to feed some data to the Kafka to reproduce this >> issue with your script? >>>> Does this OOM issue also happen when you are not using the >> Kafka source/sink? >>>> Piotrek >>>> On 8 Nov 2017, at 14:08, Javier Lopez <javier.lo...@zalando.de> >> wrote: >>>> Hi, >>>> This is the test flink job we created to trigger this leak >> https://gist.github.com/javieredo/c6052404dbe6cc602e99f4669a09f7d6 >>>> And this is the python script we are using to execute the job >> thousands of times to get the OOM problem >> https://gist.github.com/javieredo/4825324d5d5f504e27ca6c004396a107 >>>> The cluster we used for this has this configuration: >>>> Instance type: t2.large >>>> Number of workers: 2 >>>> HeapMemory: 5500 >>>> Number of task slots per node: 4 >>>> TaskMangMemFraction: 0.5 >>>> NumberOfNetworkBuffers: 2000 >>>> We have tried several things, increasing the heap, reducing the >> heap, more memory fraction, changes this value in the >> taskmanager.sh "TM_MAX_OFFHEAP_SIZE="2G"; and nothing seems to >> work. >>>> Thanks for your help. >>>> On 8 November 2017 at 13:26, ÇETİNKAYA EBRU ÇETİNKAYA EBRU >> <b20926...@cs.hacettepe.edu.tr> wrote: >>> On 2017-11-08 15:20, Piotr Nowojski wrote: >>> Hi Ebru and Javier, >>> Yes, if you could share this example job it would be helpful. >>> Ebru: could you explain in a little more details how does >> your Job(s) >>> look like? Could you post some code? If you are just using >> maps and >>> filters there shouldn’t be any network transfers involved, >> aside >>> from Source and Sink functions. >>> Piotrek >>> On 8 Nov 2017, at 12:54, ebru >> <b20926...@cs.hacettepe.edu.tr> wrote: >>> Hi Javier, >>> It would be helpful if you share your test job with us. >>> Which configurations did you try? >>> -Ebru >>> On 8 Nov 2017, at 14:43, Javier Lopez >> <javier.lo...@zalando.de> >>> wrote: >>> Hi, >>> We have been facing a similar problem. We have tried some >> different >>> configurations, as proposed in other email thread by Flavio >> and >>> Kien, but it didn't work. We have a workaround similar to >> the one >>> that Flavio has, we restart the taskmanagers once they reach >> a >>> memory threshold. We created a small test to remove all of >> our >>> dependencies and leave only flink native libraries. This >> test reads >>> data from a Kafka topic and writes it back to another topic >> in >>> Kafka. We cancel the job and start another every 5 seconds. >> After >>> ~30 minutes of doing this process, the cluster reaches the >> OS memory >>> limit and dies. >>> Currently, we have a test cluster with 8 workers and 8 task >> slots >>> per node. We have one job that uses 56 slots, and we cannot >> execute >>> that job 5 times in a row because the whole cluster dies. If >> you >>> want, we can publish our test job. >>> Regards, >>> On 8 November 2017 at 11:20, Aljoscha Krettek >> <aljos...@apache.org> >>> wrote: >>> @Nico & @Piotr Could you please have a look at this? You >> both >>> recently worked on the network stack and might be most >> familiar with >>> this. >>> On 8. Nov 2017, at 10:25, Flavio Pompermaier >> <pomperma...@okkam.it> >>> wrote: >>> We also have the same problem in production. At the moment >> the >>> solution is to restart the entire Flink cluster after every >> job.. >>> We've tried to reproduce this problem with a test (see >>> https://issues.apache.org/jira/browse/FLINK-7845 [1]) but we >> don't >>> know whether the error produced by the test and the leak are >>> correlated.. >>> Best, >>> Flavio >>> On Wed, Nov 8, 2017 at 9:51 AM, ÇETİNKAYA EBRU ÇETİNKAYA >> EBRU >>> <b20926...@cs.hacettepe.edu.tr> wrote: >>> On 2017-11-07 16:53, Ufuk Celebi wrote: >>> Do you use any windowing? If yes, could you please share >> that code? >>> If >>> there is no stateful operation at all, it's strange where >> the list >>> state instances are coming from. >>> On Tue, Nov 7, 2017 at 2:35 PM, ebru >> <b20926...@cs.hacettepe.edu.tr> >>> wrote: >>> Hi Ufuk, >>> We don’t explicitly define any state descriptor. We only >> use map >>> and filters >>> operator. We thought that gc handle clearing the flink’s >> internal >>> states. >>> So how can we manage the memory if it is always increasing? >>> - Ebru >>> On 7 Nov 2017, at 16:23, Ufuk Celebi <u...@apache.org> wrote: >>> Hey Ebru, the memory usage might be increasing as long as a >> job is >>> running. >>> This is expected (also in the case of multiple running >> jobs). The >>> screenshots are not helpful in that regard. :-( >>> What kind of stateful operations are you using? Depending on >> your >>> use case, >>> you have to manually call `clear()` on the state instance in >> order >>> to >>> release the managed state. >>> Best, >>> Ufuk >>> On Tue, Nov 7, 2017 at 12:43 PM, ebru >>> <b20926...@cs.hacettepe.edu.tr> wrote: >>> Begin forwarded message: >>> From: ebru <b20926...@cs.hacettepe.edu.tr> >>> Subject: Re: Flink memory leak >>> Date: 7 November 2017 at 14:09:17 GMT+3 >>> To: Ufuk Celebi <u...@apache.org> >>> Hi Ufuk, >>> There are there snapshots of htop output. >>> 1. snapshot is initial state. >>> 2. snapshot is after submitted one job. >>> 3. Snapshot is the output of the one job with 15000 EPS. And >> the >>> memory >>> usage is always increasing over time. >>> <1.png><2.png><3.png> >>> On 7 Nov 2017, at 13:34, Ufuk Celebi <u...@apache.org> wrote: >>> Hey Ebru, >>> let me pull in Aljoscha (CC'd) who might have an idea what's >> causing >>> this. >>> Since multiple jobs are running, it will be hard to >> understand to >>> which job the state descriptors from the heap snapshot >> belong to. >>> - Is it possible to isolate the problem and reproduce the >> behaviour >>> with only a single job? >>> – Ufuk >>> On Tue, Nov 7, 2017 at 10:27 AM, ÇETİNKAYA EBRU >> ÇETİNKAYA EBRU >>> <b20926...@cs.hacettepe.edu.tr> wrote: >>> Hi, >>> We are using Flink 1.3.1 in production, we have one job >> manager and >>> 3 task >>> managers in standalone mode. Recently, we've noticed that we >> have >>> memory >>> related problems. We use docker container to serve Flink >> cluster. We >>> have >>> 300 slots and 20 jobs are running with parallelism of 10. >> Also the >>> job >>> count >>> may be change over time. Taskmanager memory usage always >> increases. >>> After >>> job cancelation this memory usage doesn't decrease. We've >> tried to >>> investigate the problem and we've got the task manager jvm >> heap >>> snapshot. >>> According to the jam heap analysis, possible memory leak was >> Flink >>> list >>> state descriptor. But we are not sure that is the cause of >> our >>> memory >>> problem. How can we solve the problem? >>> We have two types of Flink job. One has no state full >> operator >>> contains only maps and filters and the other has time window >> with >>> count trigger. >>> * We've analysed the jvm heaps again in different >> conditions. First >>> we analysed the snapshot when no flink jobs running on >> cluster. (image >>> 1) >>> * Then, we analysed the jvm heap snapshot when the flink job >> that has >>> no state full operator is running. And according to the >> results, leak >>> suspect was NetworkBufferPool (image 2) >>> * Last analys, there were both two types of jobs running >> and leak >>> suspect was again NetworkBufferPool. (image 3) >>> In our system jobs are regularly cancelled and resubmitted so >> we >>> noticed that when job is submitted some amount of memory >> allocated and >>> after cancelation this allocated memory never freed. So over >> time >>> memory usage is always increasing and exceeded the limits. >> Links: >> ------ >> [1] https://issues.apache.org/jira/browse/FLINK-7845 >> Hi Piotr, >> There are two types of jobs. >> In first, we use Kafka source and Kafka sink, there isn't any >> window operator. >>> In second job, we use Kafka source, filesystem sink and >> elastic search sink and window operator for buffering. >> Hi Piotrek, >> Thanks for your reply. >> We've tested our link cluster again. We have 360 slots, and our >> cluster configuration is like this; >> jobmanager.rpc.address: %JOBMANAGER% >> jobmanager.rpc.port: 6123 >> jobmanager.heap.mb: 1536 >> taskmanager.heap.mb: 1536 >> taskmanager.numberOfTaskSlots: 120 >> taskmanager.memory.preallocate: false >> parallelism.default: 1 >> jobmanager.web.port: 8081 >> state.backend: filesystem >> state.backend.fs.checkpointdir: file:///storage/%CHECKPOINTDIR% >> state.checkpoints.dir: file:///storage/%CHECKPOINTDIR% >> taskmanager.network.numberOfBuffers: 5000 >> We are using docker based Flink cluster. >> WE submitted 36 jobs with the parallelism of 10. After all slots >> became full. Memory usage were increasing by the time and one by one >> task managers start to die. And the exception was like this; >> Taskmanager1 log: >> Uncaught error from thread [flink-akka.actor.default-dispatcher-17] >> shutting down JVM since 'akka.jvm-exit-on-fatal-error' is enabled for >> ActorSystem[flink] >> java.lang.NoClassDefFoundError: >> org/apache/kafka/common/metrics/stats/Rate$1 >> at >> org.apache.kafka.common.metrics.stats.Rate.convert(Rate.java:93) >> at >> org.apache.kafka.common.metrics.stats.Rate.measure(Rate.java:62) >> at >> org.apache.kafka.common.metrics.KafkaMetric.value(KafkaMetric.java:61) >> at >> org.apache.kafka.common.metrics.KafkaMetric.value(KafkaMetric.java:52) >> at >> org.apache.flink.streaming.connectors.kafka.internals.metrics.KafkaMetricWrapper.getValue(KafkaMetricWrapper.java:35) >> at >> org.apache.flink.streaming.connectors.kafka.internals.metrics.KafkaMetricWrapper.getValue(KafkaMetricWrapper.java:26) >> at >> org.apache.flink.runtime.metrics.dump.MetricDumpSerialization.serializeGauge(MetricDumpSerialization.java:213) >> at >> org.apache.flink.runtime.metrics.dump.MetricDumpSerialization.access$200(MetricDumpSerialization.java:50) >> at >> org.apache.flink.runtime.metrics.dump.MetricDumpSerialization$MetricDumpSerializer.serialize(MetricDumpSerialization.java:138) >> at >> org.apache.flink.runtime.metrics.dump.MetricQueryService.onReceive(MetricQueryService.java:109) >> at >> akka.actor.UntypedActor$$anonfun$receive$1.applyOrElse(UntypedActor.scala:167) >> at akka.actor.Actor$class.aroundReceive(Actor.scala:467) >> at akka.actor.UntypedActor.aroundReceive(UntypedActor.scala:97) >> at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516) >> at akka.actor.ActorCell.invoke(ActorCell.scala:487) >> at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:238) >> at akka.dispatch.Mailbox.run(Mailbox.scala:220) >> at >> akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:397) >> at >> scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) >> at >> scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339) >> at >> scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) >> at >> scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107) >> Caused by: java.lang.ClassNotFoundException: >> org.apache.kafka.common.metrics.stats.Rate$1 >> at java.net.URLClassLoader.findClass(URLClassLoader.java:381) >> at java.lang.ClassLoader.loadClass(ClassLoader.java:424) >> at java.lang.ClassLoader.loadClass(ClassLoader.java:357) >> ... 22 more >> Taskmanager2 log: >> Uncaught error from thread [flink-akka.actor.default-dispatcher-17] >> shutting down JVM since 'akka.jvm-exit-on-fatal-error' is enabled for >> ActorSystem[flink] >> Java.lang.NoClassDefFoundError: >> org/apache/flink/streaming/connectors/kafka/internals/AbstractFetcher$1 >> at >> org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher$OffsetGauge.getValue(AbstractFetcher.java:492) >> at >> org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher$OffsetGauge.getValue(AbstractFetcher.java:480) >> at >> org.apache.flink.runtime.metrics.dump.MetricDumpSerialization.serializeGauge(MetricDumpSerialization.java:213) >> at >> org.apache.flink.runtime.metrics.dump.MetricDumpSerialization.access$200(MetricDumpSerialization.java:50) >> at >> org.apache.flink.runtime.metrics.dump.MetricDumpSerialization$MetricDumpSerializer.serialize(MetricDumpSerialization.java:138) >> at >> org.apache.flink.runtime.metrics.dump.MetricQueryService.onReceive(MetricQueryService.java:109) >> at >> akka.actor.UntypedActor$$anonfun$receive$1.applyOrElse(UntypedActor.scala:167) >> at akka.actor.Actor$class.aroundReceive(Actor.scala:467) >> at akka.actor.UntypedActor.aroundReceive(UntypedActor.scala:97) >> at akka.actor.ActorCell.receiveMessage(ActorCell.scala:516) >> at akka.actor.ActorCell.invoke(ActorCell.scala:487) >> at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:238) >> at akka.dispatch.Mailbox.run(Mailbox.scala:220) >> at >> akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:397) >> at >> scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) >> at >> scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339) >> at >> scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) >> at >> scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107) >> Caused by: java.lang.ClassNotFoundException: >> org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher$1 >> at java.net.URLClassLoader.findClass(URLClassLoader.java:381) >> at java.lang.ClassLoader.loadClass(ClassLoader.java:424) >> at java.lang.ClassLoader.loadClass(ClassLoader.java:357) >> ... 18 more >> -Ebru > Hi Piotrek, > > We attached the full log of the taskmanager1. > This may not be a dependency issue because until all of the task slots is > full, we didn't get any No Class Def Found exception, when there is available > memory jobs can run without exception for days. > Also there is Kafka Instance Already Exist exception in full log, but this > not relevant and doesn't effect jobs or task managers. > > -Ebru<taskmanager1.log.zip>
