Affects Version/s: None
Fix Version/s: 0.10.0.0
1. Add a K-V store with format <key, window> : <agg-value> with Windowed<K> ser-de and <T> ser-de.
2. Upon receiving a record forwarded from the upstream processor:
2.1. First get the mapping window(s) from the context timestamp.
2.2. For each window, first try fetch from the store, if not exist call initialValue().
2.2. Only call add(..), and never call remove(..)
2.3. Forward the aggregate value with key Windowed<agg-key> and value Change<agg-value>.