Yep, makes sense. On Fri, Mar 12, 2021 at 10:12 AM Roman Khachatryan <ro...@apache.org> wrote:
> > Want to confirm that the keys are GCed ( along with state ) once the > (windows close + lateness ) ? > Window state is cleared (as well as the window itself), but global > state is not (unless you use TTL). > > [1] > https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/state/state.html#state-time-to-live-ttl > > Regards, > Roman > > On Fri, Mar 12, 2021 at 2:16 PM Vishal Santoshi > <vishal.santo...@gmail.com> wrote: > > > > Sometimes writing it down makes you think. I now realize that this is > not the right approach, given that merging windows will have their own > states..and how the merge happens is really at the key level.... > > > > > > > > On Fri, Mar 12, 2021 at 6:27 AM Vishal Santoshi < > vishal.santo...@gmail.com> wrote: > >> > >> I intend to augment every event in a session with a unique ID. To > keep the session lean, there is a PurgingTrigger on this aggregate that > fires on a count of 1. > >> > >> >> (except that the number of keys can grow). > >> > >> Want to confirm that the keys are GCed ( along with state ) once the > (windows close + lateness ) ? > >> > >> > >> > >> On Fri, Mar 12, 2021 at 5:32 AM Roman Khachatryan <ro...@apache.org> > wrote: > >>> > >>> Hi Vishal, > >>> > >>> There is no leak in the code you provided (except that the number of > >>> keys can grow). > >>> But as you figured out the state is scoped to key, not to window+key. > >>> > >>> Could you explain what you are trying to achieve and why do you need > to combine > >>> sliding windows with state scoped to window+key? > >>> > >>> Regards, > >>> Roman > >>> > >>> On Fri, Mar 12, 2021 at 5:13 AM Vishal Santoshi > >>> <vishal.santo...@gmail.com> wrote: > >>> > > >>> > Essentially, Does this code leak state > >>> > > >>> > private static class SessionIdProcessWindowFunction<KEY extends > java.io.Serializable, VALUE extends java.io.Serializable> > >>> > extends > >>> > ProcessWindowFunction<KeyedSession<KEY, VALUE>, > KeyedSessionWithSessionID<KEY, VALUE>, KEY, TimeWindow> { > >>> > private static final long serialVersionUID = 1L; > >>> > private final static ValueStateDescriptor<String> sessionId = new > ValueStateDescriptor<String>("session_uid", > >>> > String.class); > >>> > > >>> > @Override > >>> > public void process(KEY key, > >>> > ProcessWindowFunction<KeyedSession<KEY, VALUE>, > KeyedSessionWithSessionID<KEY, VALUE>, KEY, TimeWindow>.Context context, > >>> > Iterable<KeyedSession<KEY, VALUE>> elements, > Collector<KeyedSessionWithSessionID<KEY, VALUE>> out) > >>> > throws Exception { > >>> > // I need this scoped to key/window > >>> > if (getRuntimeContext().getState(sessionId).value() == null) { > >>> > UUID uuid = UUID.randomUUID(); > >>> > getRuntimeContext().getState(sessionId).update(uuid.toString()); > >>> > } > >>> > String uuid = getRuntimeContext().getState(sessionId).value(); > >>> > out.collect(new > KeyedSessionWithSessionID<>(elements.iterator().next(), uuid)); > >>> > } > >>> > } > >>> > > >>> > On Thu, Mar 11, 2021 at 11:09 PM Vishal Santoshi < > vishal.santo...@gmail.com> wrote: > >>> >> > >>> >> Hello folks, > >>> >> The suggestion is to use windowState() for a key > key per window state and clear the state explicitly. Also it seems that > getRuntime().getState() will return a globalWindow() where state is shared > among windows with the same key. I desire of course to have state scoped to > a key per window and was wanting to use windowState().. The caveat is that > my window is a Session Window and when I try to use clear() I am thrown > this exception ( Session Windows are Merging Windows ) > >>> >> > >>> >> Caused by: java.lang.UnsupportedOperationException: Per-window > state is not allowed when using merging windows. > >>> >> > >>> >> > >>> >> The questions are > >>> >> > >>> >> * How do I have state per session window/ per key and still be able > to clear it ? > >>> >> * Does getRuntime().getState() give me the clear() semantics for > free along with state per window per key and thus I have understood > getRuntime().getState() wrong ? > >>> >> > >>> >> Regards. > >>> >> > >>> >> > >>> >> >
