@RequiresTimeSortedInput might feed data out of order

StatefulDoFnRunner drops late data based on output watermark, but output watermark might not advance fast enough to discard data that should have been dropped. We have to either:

 a) drop data based on input watermark, or

 b) drop data based on timestamp of last emitted element

I'm a little in favor of b), because it can drop less data.