Ive added the 'until()' clause to some aggregation steps and it's working wonders for keeping the size of the state store in useful boundaries... But Im not 100% clear on how it works.
What is implied by the '.until()' clause? What determines when to stop receiving further data - is it clock time (since the window was created)? It seems problematic for it to refer to EventTime as this may bounce all over the place. For non-overlapping windows a given record can only fall into a single aggregation period - so when would a value get discarded? Im using 'groupByKey(),aggregate(..., TimeWindows.of(60 * 1000L).until(10 * 1000L))' - but what is this accomplishing?
