Flink workers OOM Stream2Batch application

[Javier Lopez <ja...@zalando.de>]

Hi all,

One of our use cases implies to do some Stream2Batch processing. We are
using Flink to read from a streaming source and deliver files to S3, after
applying some transformation to the stream. These Flink jobs are not
running 24/7, they are running on demand and consume a finite number of
records. Once the job has consumed all the records, we stop the job.

We have seen that for this particular use case the memory consumption
increases over time, leading after several hours to an instance crash with
the error "There is insufficient memory for the Java Runtime Environment to
continue.". We have monitored the heap memory and the OS memory. The heap
memory usage remains constant over time, but the OS memory keeps increasing
until the point it kills the worker.

We are using other Flink clusters for other use cases, which have jobs
running 24/7 for several months without a problem. We checked as well the
memory behavior on them and found out that both heap and OS memory usage
remain constant.

We did as well another comparison, we dumped the JVM objects to check how
many of them are we keeping in memory in each cluster. We found out that
the cluster running this stream2batch jobs keeps almost 5 times more
objects than the other cluster which has jobs running 24/7. In 4 hours the
stream2batch cluster kept 5 times more objects in memory than the other
cluster (which has been processing data constantly for 2 months).

We did this "short" introduction to our use cases because we believe that
there is a problem when we stop the Flink jobs. It looks like the memory is
not cleaned when the jobs are stopped. Is this an expected behavior?

BTW, we are using this version: FLINK_VERSION="1.3.0"
HADOOP_VERSION="hadoop27" SCALA_VERSION="scala_2.10"

Regards,