Use constant length windowing in producer
It turns out WindowStagger.NATURAL combined with the TumblingEventTimeWindow did not work as we wanted. Specifically every 5 minutes it emits all it's windows. What we want is for windows to be flushed 5 minutes after the first event of the window was recieved.
To accomplish this I started with the EventTimeSessionWindows which almost does what we need. The only change was a modification to the window merge process which merges windows when they intersect, but does not expand the length of the window. Instead the window length is constant. In testing this appears to work as necessary, with events constantly flowing out of the producer when the 5 minute windows expire.