Hi everyone, I have a stateless application (Flink 1.16.1) that consumes at a rate of about 12 records per second and uses flatmap to write n records to one (Delta Table) sink and m records to another (Delta Table) sink. My Task Manager's heap memory usage seems incredibly high for such a simple task. Does anyone know why this would be? Or how I can go about decreasing this? I would think that a stateless application would have a way smaller memory footprint. Here's a pic of the memory usage

what is your checkpoint interval?

Hi @Anmol , the checkpoint interval is 5mins for the image shown, which results in the periodic heap memory usage of 5mins. But changing this hasn't yielded a smaller maximum heap memory used. I've Experimented with everything from 5s to 10minutes. These changes have affected the profile of the memory usage, but not the maximum.

can you try profiling your job https://nightlies.apache.org/flink/flink-docs-master/docs/ops/debugging/application_profiling/ and see which class is taking too much memory. had similar issues with iceberg connector once, issue was the library was listing all the parts which were already merged, was solved in a later bug fix.

@Anmol I'll give that a go. Thanks 😄 👍

When you say that you see your TaskManager’s memory usage is high, how are you determining this? I see the graph that shows the usage, but Java’s garbage collector is lazy by design. This means the usage amount will contain lots of dead objects that could be cleaned up, but aren’t. Based on my experience, the dips that you see periodically to about ~1GB-2.5GB of usage are likely indicative of the application’s “real” usage, i.e. the total size of live objects. I suspect the dips are when the GC’s “Old Generation” is cleaned up. Since your application doesn’t need this space, the GC is probably only cleaning it up only every few minutes. If you wanted to test this in a simple way, you could simply re-run your application with progressively smaller memory amounts. It looks like it’s set to 7GB now. I’d try 4GB and run it for a bit. Then, you could lower it to 3GB, and run it again for a while. If you try 2GB, it may work — it is hard to be sure from the graph alone. 3GB is the safest option based on what I see.

Hi @David Christle, sorry only saw this now... we solved it, but we did try that ☝️ and it didn't work. Anything under 7GB would result in intermittent OOM issues and a job restart. We dug into a helluva lot of things, and ultimately it's solely due to Delta Table being the sink. Cause of the problem: It was entirely due to the number of partitions on the Delta Table. It's pretty unbelievable, but we simply removed 3 of the partitions (going from 7 to 4), and our memory usage went from hovering around 8GB to about 500MB 🤯 . The above is (at least) due to 2 reasons, the Delta Table checkpoint growing much much faster with highly partitioned data (and it gets downloaded and kept in memory constantly), and how records are kept in memory before being committed.

That’s so crazy! From the first plot, it is clear that yes the garbage collector has to do some work to keep the heap small. Hopefully the smaller number of partitions is OK for your throughput. If you do need to scale up in the future, you could try some allocation profiling or to catch a heap dump right near the peak of memory usage (can be tricky). Do you have any partition keys on your table, and if so, do you do anything to shuffle keys to the same task? (Probably not - it can be done but it’s somewhat annoying in Flink, in my experience).

@David Christle yes quite a few partition keys. It's such a simple ETL application that there's no need to do any of the shuffling you describe. Worth noting, after decreasing the partitions to 4, each of their values won't change at all until the next day as we have version/year/month/day. So it's practically 0 partitions. The memory footprint is so small now it's almost unbelievable. Even at high throughput (around 22k records ingested and 80k written to 2 delta tables per second), it barely goes over 0.8GB.