One unfortunate aspect of using a fold() instead of a window is that the fold function has no knowledge of the watermarks. As a result, it is difficult to ensure that only items before the current watermark are included in the aggregation, and that old items are evicted correctly. This fact lends more support to the idea of using a custom operator (though that is more complex) or adding support for this use case within Flink.
-Shannon
