beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From tg...@apache.org
Subject [1/2] incubator-beam git commit: Revert "Improvements to ReduceFnRunner prefetching"
Date Wed, 30 Nov 2016 23:31:58 GMT
Repository: incubator-beam
Updated Branches:
  refs/heads/master 4b682039d -> c8f2cdb22


Revert "Improvements to ReduceFnRunner prefetching"

This reverts commit 4282c67c5fa4dea2fe6c8695e0ea23f383c6457b, which
contained some incompatibilities outside of runners-core.


Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/aaa3b91e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/aaa3b91e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/aaa3b91e

Branch: refs/heads/master
Commit: aaa3b91e1e7b39dd585314a6017235cdd127e923
Parents: 4b68203
Author: Kenneth Knowles <klk@google.com>
Authored: Wed Nov 30 15:21:53 2016 -0800
Committer: Kenneth Knowles <klk@google.com>
Committed: Wed Nov 30 15:21:53 2016 -0800

----------------------------------------------------------------------
 .../core/GroupAlsoByWindowViaWindowSetDoFn.java |  11 +-
 .../beam/runners/core/PaneInfoTracker.java      |   4 -
 .../runners/core/ReduceFnContextFactory.java    |   9 +-
 .../beam/runners/core/ReduceFnRunner.java       | 488 +++++++------------
 .../apache/beam/runners/core/WatermarkHold.java |   5 -
 .../triggers/TriggerStateMachineRunner.java     |  14 +-
 .../beam/runners/core/ReduceFnTester.java       |   4 +-
 .../GroupAlsoByWindowEvaluatorFactory.java      |   5 +-
 .../apache/beam/sdk/transforms/DoFnTester.java  |   6 +-
 .../sdk/util/state/InMemoryTimerInternals.java  |  22 +-
 .../beam/sdk/util/state/TimerCallback.java      |   9 +-
 .../util/state/InMemoryTimerInternalsTest.java  |  54 +-
 12 files changed, 224 insertions(+), 407 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/aaa3b91e/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowViaWindowSetDoFn.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowViaWindowSetDoFn.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowViaWindowSetDoFn.java
index 294f21d..8b10813 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowViaWindowSetDoFn.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/GroupAlsoByWindowViaWindowSetDoFn.java
@@ -27,6 +27,7 @@ import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
 import org.apache.beam.sdk.util.KeyedWorkItem;
 import org.apache.beam.sdk.util.SystemDoFnInternal;
 import org.apache.beam.sdk.util.TimerInternals;
+import org.apache.beam.sdk.util.TimerInternals.TimerData;
 import org.apache.beam.sdk.util.WindowingStrategy;
 import org.apache.beam.sdk.util.state.StateInternals;
 import org.apache.beam.sdk.util.state.StateInternalsFactory;
@@ -72,9 +73,9 @@ public class GroupAlsoByWindowViaWindowSetDoFn<
 
   @Override
   public void processElement(ProcessContext c) throws Exception {
-    KeyedWorkItem<K, InputT> keyedWorkItem = c.element();
+    KeyedWorkItem<K, InputT> element = c.element();
 
-    K key = keyedWorkItem.key();
+    K key = c.element().key();
     TimerInternals timerInternals = c.windowingInternals().timerInternals();
     StateInternals<K> stateInternals = stateInternalsFactory.stateInternalsForKey(key);
 
@@ -92,8 +93,10 @@ public class GroupAlsoByWindowViaWindowSetDoFn<
             reduceFn,
             c.getPipelineOptions());
 
-    reduceFnRunner.processElements(keyedWorkItem.elementsIterable());
-    reduceFnRunner.onTimers(keyedWorkItem.timersIterable());
+    reduceFnRunner.processElements(element.elementsIterable());
+    for (TimerData timer : element.timersIterable()) {
+      reduceFnRunner.onTimer(timer);
+    }
     reduceFnRunner.persist();
   }
 

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/aaa3b91e/runners/core-java/src/main/java/org/apache/beam/runners/core/PaneInfoTracker.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/PaneInfoTracker.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/PaneInfoTracker.java
index 69a4cfd..8140243 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/PaneInfoTracker.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/PaneInfoTracker.java
@@ -54,10 +54,6 @@ public class PaneInfoTracker {
     state.access(PANE_INFO_TAG).clear();
   }
 
-  public void prefetchPaneInfo(ReduceFn<?, ?, ?, ?>.Context context) {
-    context.state().access(PaneInfoTracker.PANE_INFO_TAG).readLater();
-  }
-
   /**
    * Return a ({@link ReadableState} for) the pane info appropriate for {@code context}. The pane
    * info includes the timing for the pane, who's calculation is quite subtle.

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/aaa3b91e/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnContextFactory.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnContextFactory.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnContextFactory.java
index c5bda9b..539126a 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnContextFactory.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnContextFactory.java
@@ -37,6 +37,7 @@ import org.apache.beam.sdk.util.TimerInternals.TimerData;
 import org.apache.beam.sdk.util.Timers;
 import org.apache.beam.sdk.util.WindowingStrategy;
 import org.apache.beam.sdk.util.state.MergingStateAccessor;
+import org.apache.beam.sdk.util.state.ReadableState;
 import org.apache.beam.sdk.util.state.State;
 import org.apache.beam.sdk.util.state.StateAccessor;
 import org.apache.beam.sdk.util.state.StateContext;
@@ -116,7 +117,7 @@ class ReduceFnContextFactory<K, InputT, OutputT, W extends BoundedWindow> {
   }
 
   public ReduceFn<K, InputT, OutputT, W>.OnTriggerContext forTrigger(W window,
-      PaneInfo pane, StateStyle style, OnTriggerCallbacks<OutputT> callbacks) {
+      ReadableState<PaneInfo> pane, StateStyle style, OnTriggerCallbacks<OutputT> callbacks) {
     return new OnTriggerContextImpl(stateAccessor(window, style), pane, callbacks);
   }
 
@@ -388,11 +389,11 @@ class ReduceFnContextFactory<K, InputT, OutputT, W extends BoundedWindow> {
 
   private class OnTriggerContextImpl extends ReduceFn<K, InputT, OutputT, W>.OnTriggerContext {
     private final StateAccessorImpl<K, W> state;
-    private final PaneInfo pane;
+    private final ReadableState<PaneInfo> pane;
     private final OnTriggerCallbacks<OutputT> callbacks;
     private final TimersImpl timers;
 
-    private OnTriggerContextImpl(StateAccessorImpl<K, W> state, PaneInfo pane,
+    private OnTriggerContextImpl(StateAccessorImpl<K, W> state, ReadableState<PaneInfo> pane,
         OnTriggerCallbacks<OutputT> callbacks) {
       reduceFn.super();
       this.state = state;
@@ -423,7 +424,7 @@ class ReduceFnContextFactory<K, InputT, OutputT, W extends BoundedWindow> {
 
     @Override
     public PaneInfo paneInfo() {
-      return pane;
+      return pane.read();
     }
 
     @Override

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/aaa3b91e/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnRunner.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnRunner.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnRunner.java
index 3a82be9..a686f46 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnRunner.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnRunner.java
@@ -21,15 +21,12 @@ import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkState;
 
 import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Function;
-import com.google.common.collect.FluentIterable;
-import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.ImmutableMap;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
-import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -61,6 +58,7 @@ import org.apache.beam.sdk.util.WindowTracing;
 import org.apache.beam.sdk.util.WindowedValue;
 import org.apache.beam.sdk.util.WindowingStrategy;
 import org.apache.beam.sdk.util.WindowingStrategy.AccumulationMode;
+import org.apache.beam.sdk.util.state.ReadableState;
 import org.apache.beam.sdk.util.state.StateInternals;
 import org.apache.beam.sdk.util.state.StateNamespaces.WindowNamespace;
 import org.apache.beam.sdk.util.state.TimerCallback;
@@ -270,32 +268,6 @@ public class ReduceFnRunner<K, InputT, OutputT, W extends BoundedWindow> impleme
     return activeWindows.getActiveAndNewWindows().isEmpty();
   }
 
-  private Set<W> openWindows(Collection<W> windows) {
-    Set<W> result = new HashSet<>();
-    for (W window : windows) {
-      ReduceFn<K, InputT, OutputT, W>.Context directContext = contextFactory.base(
-          window, StateStyle.DIRECT);
-      if (!triggerRunner.isClosed(directContext.state())) {
-        result.add(window);
-      }
-    }
-    return result;
-  }
-
-  private Collection<W> windowsThatShouldFire(Set<W> windows) throws Exception {
-    Collection<W> result = new ArrayList<>();
-    // Filter out timers that didn't trigger.
-    for (W window : windows) {
-      ReduceFn<K, InputT, OutputT, W>.Context directContext =
-          contextFactory.base(window, StateStyle.DIRECT);
-      if (triggerRunner.shouldFire(
-          directContext.window(), directContext.timers(), directContext.state())) {
-        result.add(window);
-      }
-    }
-    return result;
-  }
-
   /**
    * Incorporate {@code values} into the underlying reduce function, and manage holds, timers,
    * triggers, and window merging.
@@ -321,54 +293,25 @@ public class ReduceFnRunner<K, InputT, OutputT, W extends BoundedWindow> impleme
    * </ol>
    */
   public void processElements(Iterable<WindowedValue<InputT>> values) throws Exception {
-    if (!values.iterator().hasNext()) {
-      return;
-    }
-
-    // Determine all the windows for elements.
-    Set<W> windows = collectWindows(values);
     // If an incoming element introduces a new window, attempt to merge it into an existing
     // window eagerly.
-    Map<W, W> windowToMergeResult = mergeWindows(windows);
-    if (!windowToMergeResult.isEmpty()) {
-      // Update windows by removing all windows that were merged away and adding
-      // the windows they were merged to. We add after completing all the
-      // removals to avoid removing a window that was also added.
-      List<W> addedWindows = new ArrayList<>(windowToMergeResult.size());
-      for (Map.Entry<W, W> entry : windowToMergeResult.entrySet()) {
-        windows.remove(entry.getKey());
-        addedWindows.add(entry.getValue());
-      }
-      windows.addAll(addedWindows);
-    }
+    Map<W, W> windowToMergeResult = collectAndMergeWindows(values);
 
-    prefetchWindowsForValues(windows);
-
-    // All windows that are open before element processing may need to fire.
-    Set<W> windowsToConsider = openWindows(windows);
+    Set<W> windowsToConsider = new HashSet<>();
 
     // Process each element, using the updated activeWindows determined by collectAndMergeWindows.
     for (WindowedValue<InputT> value : values) {
-      processElement(windowToMergeResult, value);
+      windowsToConsider.addAll(processElement(windowToMergeResult, value));
     }
 
-    // Now that we've processed the elements, see if any of the windows need to fire.
-    // Prefetch state necessary to determine if the triggers should fire.
+    // Trigger output from any window for which the trigger is ready
     for (W mergedWindow : windowsToConsider) {
-      triggerRunner.prefetchShouldFire(
-          mergedWindow, contextFactory.base(mergedWindow, StateStyle.DIRECT).state());
-    }
-    // Filter to windows that are firing.
-    Collection<W> windowsToFire = windowsThatShouldFire(windowsToConsider);
-    // Prefetch windows that are firing.
-    for (W window : windowsToFire) {
-      prefetchEmit(contextFactory.base(window, StateStyle.DIRECT),
-          contextFactory.base(window, StateStyle.RENAMED));
-    }
-    // Trigger output from firing windows.
-    for (W window : windowsToFire) {
-      emit(contextFactory.base(window, StateStyle.DIRECT),
-          contextFactory.base(window, StateStyle.RENAMED));
+      ReduceFn<K, InputT, OutputT, W>.Context directContext =
+          contextFactory.base(mergedWindow, StateStyle.DIRECT);
+      ReduceFn<K, InputT, OutputT, W>.Context renamedContext =
+          contextFactory.base(mergedWindow, StateStyle.RENAMED);
+      triggerRunner.prefetchShouldFire(mergedWindow, directContext.state());
+      emitIfAppropriate(directContext, renamedContext);
     }
 
     // We're all done with merging and emitting elements so can compress the activeWindow state.
@@ -382,61 +325,52 @@ public class ReduceFnRunner<K, InputT, OutputT, W extends BoundedWindow> impleme
   }
 
   /**
-   * Extract the windows associated with the values.
+   * Extract the windows associated with the values, and invoke merge. Return a map
+   * from windows to the merge result window. If a window is not in the domain of
+   * the result map then it did not get merged into a different window.
    */
-  private Set<W> collectWindows(Iterable<WindowedValue<InputT>> values) throws Exception {
-    Set<W> windows = new HashSet<>();
-    for (WindowedValue<?> value : values) {
-      for (BoundedWindow untypedWindow : value.getWindows()) {
-        @SuppressWarnings("unchecked")
-          W window = (W) untypedWindow;
-        windows.add(window);
-      }
-    }
-    return windows;
-  }
-
-  /**
-   * Invoke merge for the given windows and return a map from windows to the
-   * merge result window. Windows that were not merged are not present in the
-   * map.
-   */
-  private Map<W, W> mergeWindows(Set<W> windows) throws Exception {
+  private Map<W, W> collectAndMergeWindows(Iterable<WindowedValue<InputT>> values)
+      throws Exception {
+    // No-op if no merging can take place
     if (windowingStrategy.getWindowFn().isNonMerging()) {
-      // Return an empty map, indicating that every window is not merged.
-      return Collections.emptyMap();
+      return ImmutableMap.of();
     }
 
-    Map<W, W> windowToMergeResult = new HashMap<>();
     // Collect the windows from all elements (except those which are too late) and
     // make sure they are already in the active window set or are added as NEW windows.
-    for (W window : windows) {
-      // For backwards compat with pre 1.4 only.
-      // We may still have ACTIVE windows with multiple state addresses, representing
-      // a window who's state has not yet been eagerly merged.
-      // We'll go ahead and merge that state now so that we don't have to worry about
-      // this legacy case anywhere else.
-      if (activeWindows.isActive(window)) {
-        Set<W> stateAddressWindows = activeWindows.readStateAddresses(window);
-        if (stateAddressWindows.size() > 1) {
-          // This is a legacy window who's state has not been eagerly merged.
-          // Do that now.
-          ReduceFn<K, InputT, OutputT, W>.OnMergeContext premergeContext =
-              contextFactory.forPremerge(window);
-          reduceFn.onMerge(premergeContext);
-          watermarkHold.onMerge(premergeContext);
-          activeWindows.merged(window);
+    for (WindowedValue<?> value : values) {
+      for (BoundedWindow untypedWindow : value.getWindows()) {
+        @SuppressWarnings("unchecked")
+        W window = (W) untypedWindow;
+
+        // For backwards compat with pre 1.4 only.
+        // We may still have ACTIVE windows with multiple state addresses, representing
+        // a window who's state has not yet been eagerly merged.
+        // We'll go ahead and merge that state now so that we don't have to worry about
+        // this legacy case anywhere else.
+        if (activeWindows.isActive(window)) {
+          Set<W> stateAddressWindows = activeWindows.readStateAddresses(window);
+          if (stateAddressWindows.size() > 1) {
+            // This is a legacy window who's state has not been eagerly merged.
+            // Do that now.
+            ReduceFn<K, InputT, OutputT, W>.OnMergeContext premergeContext =
+                contextFactory.forPremerge(window);
+            reduceFn.onMerge(premergeContext);
+            watermarkHold.onMerge(premergeContext);
+            activeWindows.merged(window);
+          }
         }
-      }
 
-      // Add this window as NEW if it is not currently ACTIVE.
-      // If we had already seen this window and closed its trigger, then the
-      // window will not be currently ACTIVE. It will then be added as NEW here,
-      // and fall into the merging logic as usual.
-      activeWindows.ensureWindowExists(window);
+        // Add this window as NEW if it is not currently ACTIVE.
+        // If we had already seen this window and closed its trigger, then the
+        // window will not be currently ACTIVE. It will then be added as NEW here,
+        // and fall into the merging logic as usual.
+        activeWindows.ensureWindowExists(window);
+      }
     }
 
     // Merge all of the active windows and retain a mapping from source windows to result windows.
+    Map<W, W> windowToMergeResult = new HashMap<>();
     activeWindows.merge(new OnMergeCallback(windowToMergeResult));
     return windowToMergeResult;
   }
@@ -538,50 +472,38 @@ public class ReduceFnRunner<K, InputT, OutputT, W extends BoundedWindow> impleme
   }
 
   /**
-   * Redirect element windows to the ACTIVE windows they have been merged into.
-   * The compressed representation (value, {window1, window2, ...}) actually represents
-   * distinct elements (value, window1), (value, window2), ...
-   * so if window1 and window2 merge, the resulting window will contain both copies
-   * of the value.
+   * Process an element.
+   *
+   * @param value the value being processed
+   * @return the set of windows in which the element was actually processed
    */
-  private ImmutableSet<W> toMergedWindows(final Map<W, W> windowToMergeResult,
-      final Collection<? extends BoundedWindow> windows) {
-    return ImmutableSet.copyOf(
-        FluentIterable.from(windows).transform(
-            new Function<BoundedWindow, W>() {
-              @Override
-              public W apply(BoundedWindow untypedWindow) {
-                @SuppressWarnings("unchecked")
-                W window = (W) untypedWindow;
-                W mergedWindow = windowToMergeResult.get(window);
-                // If the element is not present in the map, the window is unmerged.
-                return (mergedWindow == null) ? window : mergedWindow;
-              }
-            }
-        ));
-  }
+  private Collection<W> processElement(Map<W, W> windowToMergeResult, WindowedValue<InputT> value)
+      throws Exception {
+    // Redirect element windows to the ACTIVE windows they have been merged into.
+    // The compressed representation (value, {window1, window2, ...}) actually represents
+    // distinct elements (value, window1), (value, window2), ...
+    // so if window1 and window2 merge, the resulting window will contain both copies
+    // of the value.
+    Collection<W> windows = new ArrayList<>();
+    for (BoundedWindow untypedWindow : value.getWindows()) {
+      @SuppressWarnings("unchecked")
+      W window = (W) untypedWindow;
+      W mergeResult = windowToMergeResult.get(window);
+      if (mergeResult == null) {
+        mergeResult = window;
+      }
+      windows.add(mergeResult);
+    }
 
-  private void prefetchWindowsForValues(Collection<W> windows) {
     // Prefetch in each of the windows if we're going to need to process triggers
     for (W window : windows) {
-      ReduceFn<K, InputT, OutputT, W>.Context directContext = contextFactory.base(
-          window, StateStyle.DIRECT);
+      ReduceFn<K, InputT, OutputT, W>.ProcessValueContext directContext = contextFactory.forValue(
+          window, value.getValue(), value.getTimestamp(), StateStyle.DIRECT);
       triggerRunner.prefetchForValue(window, directContext.state());
     }
-  }
-
-  /**
-   * Process an element.
-   *
-   * @param windowToMergeResult map of windows to merged windows. If a window is
-   * not present it is unmerged.
-   * @param value the value being processed
-   */
-  private void processElement(Map<W, W> windowToMergeResult, WindowedValue<InputT> value)
-      throws Exception {
-    ImmutableSet<W> windows = toMergedWindows(windowToMergeResult, value.getWindows());
 
     // Process the element for each (mergeResultWindow, not closed) window it belongs to.
+    List<W> triggerableWindows = new ArrayList<>(windows.size());
     for (W window : windows) {
       ReduceFn<K, InputT, OutputT, W>.ProcessValueContext directContext = contextFactory.forValue(
           window, value.getValue(), value.getTimestamp(), StateStyle.DIRECT);
@@ -596,6 +518,7 @@ public class ReduceFnRunner<K, InputT, OutputT, W extends BoundedWindow> impleme
         continue;
       }
 
+      triggerableWindows.add(window);
       activeWindows.ensureWindowIsActive(window);
       ReduceFn<K, InputT, OutputT, W>.ProcessValueContext renamedContext = contextFactory.forValue(
           window, value.getValue(), value.getTimestamp(), StateStyle.RENAMED);
@@ -639,152 +562,102 @@ public class ReduceFnRunner<K, InputT, OutputT, W extends BoundedWindow> impleme
       // cannot take a trigger state from firing to non-firing.
       // (We don't actually assert this since it is too slow.)
     }
+
+    return triggerableWindows;
   }
 
   /**
-   * Enriches TimerData with state necessary for processing a timer as well as
-   * common queries about a timer.
+   * Called when an end-of-window, garbage collection, or trigger-specific timer fires.
    */
-  private class EnrichedTimerData {
-    public final Instant timestamp;
-    public final ReduceFn<K, InputT, OutputT, W>.Context directContext;
-    public final ReduceFn<K, InputT, OutputT, W>.Context renamedContext;
-    // If this is an end-of-window timer then we may need to set a garbage collection timer
-    // if allowed lateness is non-zero.
-    public final boolean isEndOfWindow;
-    // If this is a garbage collection timer then we should trigger and
-    // garbage collect the window.  We'll consider any timer at or after the
-    // end-of-window time to be a signal to garbage collect.
-    public final boolean isGarbageCollection;
-
-    EnrichedTimerData(
-        TimerData timer,
-        ReduceFn<K, InputT, OutputT, W>.Context directContext,
-        ReduceFn<K, InputT, OutputT, W>.Context renamedContext) {
-      this.timestamp = timer.getTimestamp();
-      this.directContext = directContext;
-      this.renamedContext = renamedContext;
-      W window = directContext.window();
-      this.isEndOfWindow = TimeDomain.EVENT_TIME == timer.getDomain()
-          && timer.getTimestamp().equals(window.maxTimestamp());
-      Instant cleanupTime = garbageCollectionTime(window);
-      this.isGarbageCollection = !timer.getTimestamp().isBefore(cleanupTime);
-    }
+  public void onTimer(TimerData timer) throws Exception {
+    // Which window is the timer for?
+    checkArgument(timer.getNamespace() instanceof WindowNamespace,
+        "Expected timer to be in WindowNamespace, but was in %s", timer.getNamespace());
+    @SuppressWarnings("unchecked")
+    WindowNamespace<W> windowNamespace = (WindowNamespace<W>) timer.getNamespace();
+    W window = windowNamespace.getWindow();
+    ReduceFn<K, InputT, OutputT, W>.Context directContext =
+        contextFactory.base(window, StateStyle.DIRECT);
+    ReduceFn<K, InputT, OutputT, W>.Context renamedContext =
+        contextFactory.base(window, StateStyle.RENAMED);
 
     // Has this window had its trigger finish?
     // - The trigger may implement isClosed as constant false.
     // - If the window function does not support windowing then all windows will be considered
     // active.
     // So we must take conjunction of activeWindows and triggerRunner state.
-    public boolean windowIsActiveAndOpen() {
-      return activeWindows.isActive(directContext.window())
-          && !triggerRunner.isClosed(directContext.state());
-    }
-  }
+    boolean windowIsActiveAndOpen =
+        activeWindows.isActive(window) && !triggerRunner.isClosed(directContext.state());
 
-  public void onTimers(Iterable<TimerData> timers) throws Exception {
-    if (!timers.iterator().hasNext()) {
-      return;
+    if (!windowIsActiveAndOpen) {
+      WindowTracing.debug(
+          "ReduceFnRunner.onTimer: Note that timer {} is for non-ACTIVE window {}", timer, window);
     }
 
-    // Create a reusable context for each timer and begin prefetching necessary
-    // state.
-    List<EnrichedTimerData> enrichedTimers = new LinkedList();
-    for (TimerData timer : timers) {
-      checkArgument(timer.getNamespace() instanceof WindowNamespace,
-          "Expected timer to be in WindowNamespace, but was in %s", timer.getNamespace());
-      @SuppressWarnings("unchecked")
-        WindowNamespace<W> windowNamespace = (WindowNamespace<W>) timer.getNamespace();
-      W window = windowNamespace.getWindow();
-      ReduceFn<K, InputT, OutputT, W>.Context directContext =
-          contextFactory.base(window, StateStyle.DIRECT);
-      ReduceFn<K, InputT, OutputT, W>.Context renamedContext =
-          contextFactory.base(window, StateStyle.RENAMED);
-      EnrichedTimerData enrichedTimer = new EnrichedTimerData(timer, directContext, renamedContext);
-      enrichedTimers.add(enrichedTimer);
-
-      // Perform prefetching of state to determine if the trigger should fire.
-      if (enrichedTimer.isGarbageCollection) {
-        triggerRunner.prefetchIsClosed(directContext.state());
-      } else {
-        triggerRunner.prefetchShouldFire(directContext.window(), directContext.state());
+    // If this is an end-of-window timer then we may need to set a garbage collection timer
+    // if allowed lateness is non-zero.
+    boolean isEndOfWindow = TimeDomain.EVENT_TIME == timer.getDomain()
+        && timer.getTimestamp().equals(window.maxTimestamp());
+
+    // If this is a garbage collection timer then we should trigger and garbage collect the window.
+    // We'll consider any timer at or after the end-of-window time to be a signal to garbage
+    // collect.
+    Instant cleanupTime = garbageCollectionTime(window);
+    boolean isGarbageCollection = TimeDomain.EVENT_TIME == timer.getDomain()
+        && !timer.getTimestamp().isBefore(cleanupTime);
+
+    if (isGarbageCollection) {
+      WindowTracing.debug(
+          "ReduceFnRunner.onTimer: Cleaning up for key:{}; window:{} at {} with "
+          + "inputWatermark:{}; outputWatermark:{}",
+          key, window, timer.getTimestamp(), timerInternals.currentInputWatermarkTime(),
+          timerInternals.currentOutputWatermarkTime());
+
+      if (windowIsActiveAndOpen) {
+        // We need to call onTrigger to emit the final pane if required.
+        // The final pane *may* be ON_TIME if no prior ON_TIME pane has been emitted,
+        // and the watermark has passed the end of the window.
+        @Nullable Instant newHold =
+            onTrigger(directContext, renamedContext, true/* isFinished */, isEndOfWindow);
+        checkState(newHold == null,
+            "Hold placed at %s despite isFinished being true.", newHold);
       }
-    }
 
-    // For those windows that are active and open, prefetch the triggering or emitting state.
-    for (EnrichedTimerData timer : enrichedTimers) {
-      if (timer.windowIsActiveAndOpen()) {
-        ReduceFn<K, InputT, OutputT, W>.Context directContext = timer.directContext;
-        if (timer.isGarbageCollection) {
-          prefetchOnTrigger(directContext, timer.renamedContext);
-        } else if (triggerRunner.shouldFire(
-            directContext.window(), directContext.timers(), directContext.state())) {
-          prefetchEmit(directContext, timer.renamedContext);
-        }
+      // Cleanup flavor B: Clear all the remaining state for this window since we'll never
+      // see elements for it again.
+      clearAllState(directContext, renamedContext, windowIsActiveAndOpen);
+    } else {
+      WindowTracing.debug(
+          "ReduceFnRunner.onTimer: Triggering for key:{}; window:{} at {} with "
+          + "inputWatermark:{}; outputWatermark:{}",
+          key, window, timer.getTimestamp(), timerInternals.currentInputWatermarkTime(),
+          timerInternals.currentOutputWatermarkTime());
+      if (windowIsActiveAndOpen) {
+        emitIfAppropriate(directContext, renamedContext);
       }
-    }
-
-    // Perform processing now that everything is prefetched.
-    for (EnrichedTimerData timer : enrichedTimers) {
-      ReduceFn<K, InputT, OutputT, W>.Context directContext = timer.directContext;
-      ReduceFn<K, InputT, OutputT, W>.Context renamedContext = timer.renamedContext;
 
-      if (timer.isGarbageCollection) {
-        WindowTracing.debug("ReduceFnRunner.onTimer: Cleaning up for key:{}; window:{} at {} with "
-                + "inputWatermark:{}; outputWatermark:{}",
-            key, directContext.window(), timer.timestamp,
-            timerInternals.currentInputWatermarkTime(),
-            timerInternals.currentOutputWatermarkTime());
-
-        boolean windowIsActiveAndOpen = timer.windowIsActiveAndOpen();
-        if (windowIsActiveAndOpen) {
-          // We need to call onTrigger to emit the final pane if required.
-          // The final pane *may* be ON_TIME if no prior ON_TIME pane has been emitted,
-          // and the watermark has passed the end of the window.
-          @Nullable
-          Instant newHold = onTrigger(
-              directContext, renamedContext, true /* isFinished */, timer.isEndOfWindow);
-          checkState(newHold == null, "Hold placed at %s despite isFinished being true.", newHold);
-        }
-
-        // Cleanup flavor B: Clear all the remaining state for this window since we'll never
-        // see elements for it again.
-        clearAllState(directContext, renamedContext, windowIsActiveAndOpen);
-      } else {
-        WindowTracing.debug("ReduceFnRunner.onTimer: Triggering for key:{}; window:{} at {} with "
-                + "inputWatermark:{}; outputWatermark:{}",
-            key, directContext.window(), timer.timestamp,
-            timerInternals.currentInputWatermarkTime(),
+      if (isEndOfWindow) {
+        // If the window strategy trigger includes a watermark trigger then at this point
+        // there should be no data holds, either because we'd already cleared them on an
+        // earlier onTrigger, or because we just cleared them on the above emitIfAppropriate.
+        // We could assert this but it is very expensive.
+
+        // Since we are processing an on-time firing we should schedule the garbage collection
+        // timer. (If getAllowedLateness is zero then the timer event will be considered a
+        // cleanup event and handled by the above).
+        // Note we must do this even if the trigger is finished so that we are sure to cleanup
+        // any final trigger finished bits.
+        checkState(
+            windowingStrategy.getAllowedLateness().isLongerThan(Duration.ZERO),
+            "Unexpected zero getAllowedLateness");
+        WindowTracing.debug(
+            "ReduceFnRunner.onTimer: Scheduling cleanup timer for key:{}; window:{} at {} with "
+            + "inputWatermark:{}; outputWatermark:{}",
+            key, directContext.window(), cleanupTime, timerInternals.currentInputWatermarkTime(),
             timerInternals.currentOutputWatermarkTime());
-        if (timer.windowIsActiveAndOpen()
-            && triggerRunner.shouldFire(
-                   directContext.window(), directContext.timers(), directContext.state())) {
-          emit(directContext, renamedContext);
-        }
-
-        if (timer.isEndOfWindow) {
-          // If the window strategy trigger includes a watermark trigger then at this point
-          // there should be no data holds, either because we'd already cleared them on an
-          // earlier onTrigger, or because we just cleared them on the above emit.
-          // We could assert this but it is very expensive.
-
-          // Since we are processing an on-time firing we should schedule the garbage collection
-          // timer. (If getAllowedLateness is zero then the timer event will be considered a
-          // cleanup event and handled by the above).
-          // Note we must do this even if the trigger is finished so that we are sure to cleanup
-          // any final trigger finished bits.
-          checkState(windowingStrategy.getAllowedLateness().isLongerThan(Duration.ZERO),
-              "Unexpected zero getAllowedLateness");
-          Instant cleanupTime = garbageCollectionTime(directContext.window());
-          WindowTracing.debug(
-              "ReduceFnRunner.onTimer: Scheduling cleanup timer for key:{}; window:{} at {} with "
-                  + "inputWatermark:{}; outputWatermark:{}",
-              key, directContext.window(), cleanupTime, timerInternals.currentInputWatermarkTime(),
-              timerInternals.currentOutputWatermarkTime());
-          checkState(!cleanupTime.isAfter(BoundedWindow.TIMESTAMP_MAX_VALUE),
-              "Cleanup time %s is beyond end-of-time", cleanupTime);
-          directContext.timers().setTimer(cleanupTime, TimeDomain.EVENT_TIME);
-        }
+        checkState(!cleanupTime.isAfter(BoundedWindow.TIMESTAMP_MAX_VALUE),
+                                 "Cleanup time %s is beyond end-of-time", cleanupTime);
+        directContext.timers().setTimer(cleanupTime, TimeDomain.EVENT_TIME);
       }
     }
   }
@@ -793,7 +666,7 @@ public class ReduceFnRunner<K, InputT, OutputT, W extends BoundedWindow> impleme
    * Clear all the state associated with {@code context}'s window.
    * Should only be invoked if we know all future elements for this window will be considered
    * beyond allowed lateness.
-   * This is a superset of the clearing done by {@link #emitPane} below since:
+   * This is a superset of the clearing done by {@link #emitIfAppropriate} below since:
    * <ol>
    * <li>We can clear the trigger finished bits since we'll never need to ask if the trigger is
    * closed again.
@@ -819,10 +692,10 @@ public class ReduceFnRunner<K, InputT, OutputT, W extends BoundedWindow> impleme
     } else {
       // If !windowIsActiveAndOpen then !activeWindows.isActive (1) or triggerRunner.isClosed (2).
       // For (1), if !activeWindows.isActive then the window must be merging and has been
-      // explicitly removed by emit. But in that case the trigger must have fired
+      // explicitly removed by emitIfAppropriate. But in that case the trigger must have fired
       // and been closed, so this case reduces to (2).
       // For (2), if triggerRunner.isClosed then the trigger was fired and entered the
-      // closed state. In that case emit will have cleared all state in
+      // closed state. In that case emitIfAppropriate will have cleared all state in
       // reduceFn, triggerRunner (except for finished bits), paneInfoTracker and activeWindows.
       // We also know nonEmptyPanes must have been unconditionally cleared by the trigger.
       // Since the trigger fired the existing watermark holds must have been cleared, and since
@@ -864,23 +737,17 @@ public class ReduceFnRunner<K, InputT, OutputT, W extends BoundedWindow> impleme
     return false;
   }
 
-  private void prefetchEmit(ReduceFn<K, InputT, OutputT, W>.Context directContext,
-                                ReduceFn<K, InputT, OutputT, W>.Context renamedContext) {
-    triggerRunner.prefetchShouldFire(directContext.window(), directContext.state());
-    triggerRunner.prefetchOnFire(directContext.window(), directContext.state());
-    triggerRunner.prefetchIsClosed(directContext.state());
-    prefetchOnTrigger(directContext, renamedContext);
-  }
-
   /**
-   * Emit if a trigger is ready to fire or timers require it, and cleanup state.
+   * Possibly emit a pane if a trigger is ready to fire or timers require it, and cleanup state.
    */
-  private void emit(
-      ReduceFn<K, InputT, OutputT, W>.Context directContext,
+  private void emitIfAppropriate(ReduceFn<K, InputT, OutputT, W>.Context directContext,
       ReduceFn<K, InputT, OutputT, W>.Context renamedContext)
       throws Exception {
-    checkState(triggerRunner.shouldFire(
-        directContext.window(), directContext.timers(), directContext.state()));
+    if (!triggerRunner.shouldFire(
+        directContext.window(), directContext.timers(), directContext.state())) {
+      // Ignore unless trigger is ready to fire
+      return;
+    }
 
     // Inform the trigger of the transition to see if it is finished
     triggerRunner.onFire(directContext.window(), directContext.timers(), directContext.state());
@@ -915,7 +782,7 @@ public class ReduceFnRunner<K, InputT, OutputT, W extends BoundedWindow> impleme
   }
 
   /**
-   * Do we need to emit?
+   * Do we need to emit a pane?
    */
   private boolean needToEmit(boolean isEmpty, boolean isFinished, PaneInfo.Timing timing) {
     if (!isEmpty) {
@@ -933,15 +800,6 @@ public class ReduceFnRunner<K, InputT, OutputT, W extends BoundedWindow> impleme
     return false;
   }
 
-  private void prefetchOnTrigger(
-      final ReduceFn<K, InputT, OutputT, W>.Context directContext,
-      ReduceFn<K, InputT, OutputT, W>.Context renamedContext) {
-    paneInfoTracker.prefetchPaneInfo(directContext);
-    watermarkHold.prefetchExtract(renamedContext);
-    nonEmptyPanes.isEmpty(renamedContext.state()).readLater();
-    reduceFn.prefetchOnTrigger(directContext.state());
-  }
-
   /**
    * Run the {@link ReduceFn#onTrigger} method and produce any necessary output.
    *
@@ -955,17 +813,25 @@ public class ReduceFnRunner<K, InputT, OutputT, W extends BoundedWindow> impleme
           throws Exception {
     Instant inputWM = timerInternals.currentInputWatermarkTime();
 
-    // Calculate the pane info.
-    final PaneInfo pane = paneInfoTracker.getNextPaneInfo(directContext, isFinished).read();
+    // Prefetch necessary states
+    ReadableState<WatermarkHold.OldAndNewHolds> outputTimestampFuture =
+        watermarkHold.extractAndRelease(renamedContext, isFinished).readLater();
+    ReadableState<PaneInfo> paneFuture =
+        paneInfoTracker.getNextPaneInfo(directContext, isFinished).readLater();
+    ReadableState<Boolean> isEmptyFuture =
+        nonEmptyPanes.isEmpty(renamedContext.state()).readLater();
 
+    reduceFn.prefetchOnTrigger(directContext.state());
+    triggerRunner.prefetchOnFire(directContext.window(), directContext.state());
+
+    // Calculate the pane info.
+    final PaneInfo pane = paneFuture.read();
     // Extract the window hold, and as a side effect clear it.
-    final WatermarkHold.OldAndNewHolds pair =
-        watermarkHold.extractAndRelease(renamedContext, isFinished).read();
+
+    WatermarkHold.OldAndNewHolds pair = outputTimestampFuture.read();
     final Instant outputTimestamp = pair.oldHold;
     @Nullable Instant newHold = pair.newHold;
 
-    final boolean isEmpty = nonEmptyPanes.isEmpty(renamedContext.state()).read();
-
     if (newHold != null) {
       // We can't be finished yet.
       checkState(
@@ -997,11 +863,11 @@ public class ReduceFnRunner<K, InputT, OutputT, W extends BoundedWindow> impleme
     }
 
     // Only emit a pane if it has data or empty panes are observable.
-    if (needToEmit(isEmpty, isFinished, pane.getTiming())) {
+    if (needToEmit(isEmptyFuture.read(), isFinished, pane.getTiming())) {
       // Run reduceFn.onTrigger method.
       final List<W> windows = Collections.singletonList(directContext.window());
       ReduceFn<K, InputT, OutputT, W>.OnTriggerContext renamedTriggerContext =
-          contextFactory.forTrigger(directContext.window(), pane, StateStyle.RENAMED,
+          contextFactory.forTrigger(directContext.window(), paneFuture, StateStyle.RENAMED,
               new OnTriggerCallbacks<OutputT>() {
                 @Override
                 public void output(OutputT toOutput) {

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/aaa3b91e/runners/core-java/src/main/java/org/apache/beam/runners/core/WatermarkHold.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/WatermarkHold.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/WatermarkHold.java
index 7f1afcc..3c04571 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/WatermarkHold.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/WatermarkHold.java
@@ -444,11 +444,6 @@ class WatermarkHold<W extends BoundedWindow> implements Serializable {
     }
   }
 
-  public void prefetchExtract(final ReduceFn<?, ?, ?, W>.Context context) {
-    context.state().access(elementHoldTag).readLater();
-    context.state().access(EXTRA_HOLD_TAG).readLater();
-  }
-
   /**
    * Return (a future for) the earliest hold for {@code context}. Clear all the holds after
    * reading, but add/restore an end-of-window or garbage collection hold if required.

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/aaa3b91e/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachineRunner.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachineRunner.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachineRunner.java
index 2f277eb..9f03216 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachineRunner.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/triggers/TriggerStateMachineRunner.java
@@ -99,25 +99,25 @@ public class TriggerStateMachineRunner<W extends BoundedWindow> {
     return readFinishedBits(state.access(FINISHED_BITS_TAG)).isFinished(rootTrigger);
   }
 
-  public void prefetchIsClosed(StateAccessor<?> state) {
+  public void prefetchForValue(W window, StateAccessor<?> state) {
     if (isFinishedSetNeeded()) {
       state.access(FINISHED_BITS_TAG).readLater();
     }
-  }
-
-  public void prefetchForValue(W window, StateAccessor<?> state) {
-    prefetchIsClosed(state);
     rootTrigger.getSpec().prefetchOnElement(
         contextFactory.createStateAccessor(window, rootTrigger));
   }
 
   public void prefetchOnFire(W window, StateAccessor<?> state) {
-    prefetchIsClosed(state);
+    if (isFinishedSetNeeded()) {
+      state.access(FINISHED_BITS_TAG).readLater();
+    }
     rootTrigger.getSpec().prefetchOnFire(contextFactory.createStateAccessor(window, rootTrigger));
   }
 
   public void prefetchShouldFire(W window, StateAccessor<?> state) {
-    prefetchIsClosed(state);
+    if (isFinishedSetNeeded()) {
+      state.access(FINISHED_BITS_TAG).readLater();
+    }
     rootTrigger.getSpec().prefetchShouldFire(
         contextFactory.createStateAccessor(window, rootTrigger));
   }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/aaa3b91e/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java
----------------------------------------------------------------------
diff --git a/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java b/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java
index 8be8ae5..337be23 100644
--- a/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java
+++ b/runners/core-java/src/test/java/org/apache/beam/runners/core/ReduceFnTester.java
@@ -509,10 +509,8 @@ public class ReduceFnTester<InputT, OutputT, W extends BoundedWindow> {
 
   public void fireTimer(W window, Instant timestamp, TimeDomain domain) throws Exception {
     ReduceFnRunner<String, InputT, OutputT, W> runner = createRunner();
-    ArrayList timers = new ArrayList(1);
-    timers.add(
+    runner.onTimer(
         TimerData.of(StateNamespaces.window(windowFn.windowCoder(), window), timestamp, domain));
-    runner.onTimers(timers);
     runner.persist();
   }
 

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/aaa3b91e/runners/direct-java/src/main/java/org/apache/beam/runners/direct/GroupAlsoByWindowEvaluatorFactory.java
----------------------------------------------------------------------
diff --git a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/GroupAlsoByWindowEvaluatorFactory.java b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/GroupAlsoByWindowEvaluatorFactory.java
index f70fb94..9d25bc6 100644
--- a/runners/direct-java/src/main/java/org/apache/beam/runners/direct/GroupAlsoByWindowEvaluatorFactory.java
+++ b/runners/direct-java/src/main/java/org/apache/beam/runners/direct/GroupAlsoByWindowEvaluatorFactory.java
@@ -47,6 +47,7 @@ import org.apache.beam.sdk.transforms.windowing.PaneInfo;
 import org.apache.beam.sdk.util.KeyedWorkItem;
 import org.apache.beam.sdk.util.SideInputReader;
 import org.apache.beam.sdk.util.TimerInternals;
+import org.apache.beam.sdk.util.TimerInternals.TimerData;
 import org.apache.beam.sdk.util.WindowTracing;
 import org.apache.beam.sdk.util.WindowedValue;
 import org.apache.beam.sdk.util.WindowingStrategy;
@@ -200,7 +201,9 @@ class GroupAlsoByWindowEvaluatorFactory implements TransformEvaluatorFactory {
       // Drop any elements within expired windows
       reduceFnRunner.processElements(
           dropExpiredWindows(key, workItem.elementsIterable(), timerInternals));
-      reduceFnRunner.onTimers(workItem.timersIterable());
+      for (TimerData timer : workItem.timersIterable()) {
+        reduceFnRunner.onTimer(timer);
+      }
       reduceFnRunner.persist();
     }
 

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/aaa3b91e/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java
index f8b1222..daa8a06 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFnTester.java
@@ -466,10 +466,8 @@ public class DoFnTester<InputT, OutputT> implements AutoCloseable {
   private static TimerCallback collectInto(final List<TimerInternals.TimerData> firedTimers) {
     return new TimerCallback() {
       @Override
-      public void onTimers(Iterable<TimerInternals.TimerData> timers) throws Exception {
-        for (TimerInternals.TimerData timer : timers) {
-          firedTimers.add(timer);
-        }
+      public void onTimer(TimerInternals.TimerData timer) throws Exception {
+        firedTimers.add(timer);
       }
     };
   }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/aaa3b91e/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/InMemoryTimerInternals.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/InMemoryTimerInternals.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/InMemoryTimerInternals.java
index f1ddaac..a3bb45a 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/InMemoryTimerInternals.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/InMemoryTimerInternals.java
@@ -21,7 +21,6 @@ import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.base.Preconditions.checkState;
 
 import com.google.common.base.MoreObjects;
-import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.PriorityQueue;
 import java.util.Set;
@@ -236,20 +235,13 @@ public class InMemoryTimerInternals implements TimerInternals {
       throws Exception {
     checkNotNull(timerCallback);
     PriorityQueue<TimerData> queue = queue(domain);
-    while (true) {
-      ArrayList<TimerData> firedTimers = new ArrayList();
-      while (!queue.isEmpty() && currentTime.isAfter(queue.peek().getTimestamp())) {
-        // Remove before firing, so that if the callback adds another identical
-        // timer we don't remove it.
-        TimerData timer = queue.remove();
-        firedTimers.add(timer);
-        WindowTracing.trace(
-            "InMemoryTimerInternals.advanceAndFire: firing {} at {}", timer, currentTime);
-      }
-      if (firedTimers.isEmpty()) {
-        break;
-      }
-      timerCallback.onTimers(firedTimers);
+    while (!queue.isEmpty() && currentTime.isAfter(queue.peek().getTimestamp())) {
+      // Remove before firing, so that if the callback adds another identical
+      // timer we don't remove it.
+      TimerData timer = queue.remove();
+      WindowTracing.trace(
+          "InMemoryTimerInternals.advanceAndFire: firing {} at {}", timer, currentTime);
+      timerCallback.onTimer(timer);
     }
   }
 }

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/aaa3b91e/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/TimerCallback.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/TimerCallback.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/TimerCallback.java
index dfdfd5b..6598e30 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/TimerCallback.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/state/TimerCallback.java
@@ -19,17 +19,16 @@ package org.apache.beam.sdk.util.state;
 
 import org.apache.beam.sdk.util.TimerInternals;
 
-
 /**
- * A callback that processes an Iterable of {@link TimerInternals.TimerData TimerData}.
+ * A callback that processes a {@link TimerInternals.TimerData TimerData}.
  */
 public interface TimerCallback {
-  /** Processes an Iterable of {@link TimerInternals.TimerData TimerData}. */
-  void onTimers(Iterable<TimerInternals.TimerData> timers) throws Exception;
+  /** Processes the {@link TimerInternals.TimerData TimerData}. */
+  void onTimer(TimerInternals.TimerData timer) throws Exception;
 
   TimerCallback NO_OP = new TimerCallback() {
     @Override
-    public void onTimers(Iterable<TimerInternals.TimerData> timers) throws Exception {
+    public void onTimer(TimerInternals.TimerData timer) throws Exception {
       // Nothing
     }
   };

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/aaa3b91e/sdks/java/core/src/test/java/org/apache/beam/sdk/util/state/InMemoryTimerInternalsTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/state/InMemoryTimerInternalsTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/state/InMemoryTimerInternalsTest.java
index a3a7749..951803a 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/state/InMemoryTimerInternalsTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/state/InMemoryTimerInternalsTest.java
@@ -17,11 +17,6 @@
  */
 package org.apache.beam.sdk.util.state;
 
-import static org.mockito.Matchers.argThat;
-
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
 import org.apache.beam.sdk.util.TimeDomain;
 import org.apache.beam.sdk.util.TimerInternals.TimerData;
 import org.joda.time.Instant;
@@ -29,7 +24,6 @@ import org.junit.Before;
 import org.junit.Test;
 import org.junit.runner.RunWith;
 import org.junit.runners.JUnit4;
-import org.mockito.ArgumentMatcher;
 import org.mockito.Mock;
 import org.mockito.Mockito;
 import org.mockito.MockitoAnnotations;
@@ -50,37 +44,6 @@ public class InMemoryTimerInternalsTest {
     MockitoAnnotations.initMocks(this);
   }
 
-  private static class TimersAre extends ArgumentMatcher<Iterable<TimerData>> {
-    final List<TimerData> expectedTimers;
-    TimersAre(List<TimerData> timers) {
-      expectedTimers = timers;
-    }
-
-    @Override
-    public boolean matches(Object actual) {
-      if (actual == null || !(actual instanceof Iterable)) {
-        return false;
-      }
-      @SuppressWarnings("unchecked")
-      Iterable<TimerData> timers = (Iterable<TimerData>) actual;
-
-      List<TimerData> actualTimers = new ArrayList();
-      for (TimerData timer : timers) {
-        actualTimers.add(timer);
-      }
-      return expectedTimers.equals(actualTimers);
-    }
-
-    @Override
-    public String toString() {
-      return "ordered timers " + expectedTimers.toString();
-    }
-  }
-
-  private static TimersAre timersAre(TimerData... timers) {
-    return new TimersAre(Arrays.asList(timers));
-  }
-
   @Test
   public void testFiringTimers() throws Exception {
     InMemoryTimerInternals underTest = new InMemoryTimerInternals();
@@ -91,7 +54,7 @@ public class InMemoryTimerInternalsTest {
     underTest.setTimer(processingTime2);
 
     underTest.advanceProcessingTime(timerCallback, new Instant(20));
-    Mockito.verify(timerCallback).onTimers(argThat(timersAre(processingTime1)));
+    Mockito.verify(timerCallback).onTimer(processingTime1);
     Mockito.verifyNoMoreInteractions(timerCallback);
 
     // Advancing just a little shouldn't refire
@@ -100,13 +63,13 @@ public class InMemoryTimerInternalsTest {
 
     // Adding the timer and advancing a little should refire
     underTest.setTimer(processingTime1);
-    Mockito.verify(timerCallback).onTimers(argThat(timersAre(processingTime1)));
+    Mockito.verify(timerCallback).onTimer(processingTime1);
     underTest.advanceProcessingTime(timerCallback, new Instant(21));
     Mockito.verifyNoMoreInteractions(timerCallback);
 
     // And advancing the rest of the way should still have the other timer
     underTest.advanceProcessingTime(timerCallback, new Instant(30));
-    Mockito.verify(timerCallback).onTimers(argThat(timersAre(processingTime2)));
+    Mockito.verify(timerCallback).onTimer(processingTime2);
     Mockito.verifyNoMoreInteractions(timerCallback);
   }
 
@@ -124,11 +87,13 @@ public class InMemoryTimerInternalsTest {
     underTest.setTimer(watermarkTime2);
 
     underTest.advanceInputWatermark(timerCallback, new Instant(30));
-    Mockito.verify(timerCallback).onTimers(argThat(timersAre(watermarkTime1, watermarkTime2)));
+    Mockito.verify(timerCallback).onTimer(watermarkTime1);
+    Mockito.verify(timerCallback).onTimer(watermarkTime2);
     Mockito.verifyNoMoreInteractions(timerCallback);
 
     underTest.advanceProcessingTime(timerCallback, new Instant(30));
-    Mockito.verify(timerCallback).onTimers(argThat(timersAre(processingTime1, processingTime2)));
+    Mockito.verify(timerCallback).onTimer(processingTime1);
+    Mockito.verify(timerCallback).onTimer(processingTime2);
     Mockito.verifyNoMoreInteractions(timerCallback);
   }
 
@@ -142,9 +107,10 @@ public class InMemoryTimerInternalsTest {
     underTest.setTimer(processingTime);
     underTest.setTimer(processingTime);
     underTest.advanceProcessingTime(timerCallback, new Instant(20));
-    Mockito.verify(timerCallback).onTimers(argThat(timersAre(processingTime)));
     underTest.advanceInputWatermark(timerCallback, new Instant(20));
-    Mockito.verify(timerCallback).onTimers(argThat(timersAre(watermarkTime)));
+
+    Mockito.verify(timerCallback).onTimer(processingTime);
+    Mockito.verify(timerCallback).onTimer(watermarkTime);
     Mockito.verifyNoMoreInteractions(timerCallback);
   }
 }


Mime
View raw message