Re: Best way to clean-up states in memory

2018-05-16 Thread Fabian Hueske
Hi Ashish, I had a look at your Trigger and couldn't spot anything that would explain leaking state. You're properly cleaning up in clear(). However, I might have found the problem for the increasing state size. A window is only completely deleted when the time passes its end timestamp (Window.ma

Re: Best way to clean-up states in memory

2018-05-14 Thread ashish pok
Thanks Fabian, Kostas, Here is what I had in the Trigger - idea is to run bitwise OR until a threshold is reached or a timeout is reached (nothing too fancy here). Let me know what you guys think. Like I said, I moved this logic to Process Function and I haven't seen the same issue I was with t

Re: Best way to clean-up states in memory

2018-05-14 Thread Fabian Hueske
Hi Ashish, Did you use per-window state (also called partitioned state) in your Trigger? If yes, you need to make sure that it is completely removed in the clear() method because processing time timers won't fire once a window was purged. So you cannot (fully) rely on timers to clean up per-window

Re: Best way to clean-up states in memory

2018-05-14 Thread Kostas Kloudas
Hi Ashish, It would be helpful to share the code of your custom trigger for the first case. Without that, we cannot tell what state you create and how/when you update/clear it. Cheers, Kostas > On May 14, 2018, at 1:04 AM, ashish pok wrote: > > Hi Till, > > Thanks for getting back. I am sure

Re: Best way to clean-up states in memory

2018-05-13 Thread ashish pok
Hi Till, Thanks for getting back. I am sure that will fix the issue but I feel like that would potentially mask an issue. I have been going back and forth with Fabian on a use case where for some of our highly transient datasets, it might make sense to just use memory based state (except of cou

Re: Best way to clean-up states in memory

2018-05-13 Thread Till Rohrmann
Hi Ashish, have you tried using Flink's RocksDBStateBackend? If your job accumulates state exceeding the available main memory, then you have to use a state backend which can spill to disk. The RocksDBStateBackend offers you exactly this functionality. Cheers, Till On Mon, Apr 30, 2018 at 3:54 P

Best way to clean-up states in memory

2018-04-30 Thread ashish pok
All, I am using noticing heap utilization creeping up slowly in couple of apps which eventually lead to OOM issue. Apps only have 1 process function that cache state. I did make sure I have a clear method invoked when events are collected normally, on exception and on timeout. Are any other best