Improve ReduceFnRunner prefetching

[scwhittle]

• add prefetch* methods for prefetching state matching existing methods
• prefetch triggers in processElements
• replace onTimer with batched onTimers method to allow prefetching across timers

[scwhittle]

@kennknowles This is a backport of changes made to beam to improve prefetching