Hello, I'm flatmapping a record into multiple "jobs" which I want to process then in parallel. Then once all the processing has completed I'd like to wait for all the results for those "jobs" and generate a single result record at the end. I'm basically trying to implement a simple scatter-gather. Given that the number of jobs depends on the original record, I was thinking about passing that total number of jobs as part of the payload of each "job", such that a keyed window operator could count in a state how many completed "jobs" were seen so far and if the count was equals to the expected total I could consider the window "complete". Which keyed window operator would you recommend to use, assuming I want to consider the window complete either when the total number of "jobs" is reached or a timeout of 10 seconds (whichever comes first)?