fix npe in ReduceRunner.

Author: stankiewicz

runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnRunner.java

@@ -376,7 +376,7 @@ public void processElements(Iterable<WindowedValue<InputT>> values) throws Excep
       emit(
           contextFactory.base(window, StateStyle.DIRECT),
           contextFactory.base(window, StateStyle.RENAMED),
-          null);
+          CausedByDrain.NORMAL);
     }
 
     // We're all done with merging and emitting elements so can compress the activeWindow state.

runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java

@@ -445,11 +445,13 @@ public <T> void output(TupleTag<T> tag, T output) {
 
     @Override
     public void outputWindowedValue(WindowedValue<OutputT> windowedValue) {
+      checkTimestamp(elem.getTimestamp(), windowedValue.getTimestamp());
       SimpleDoFnRunner.this.outputWindowedValue(mainOutputTag, windowedValue);
     }
 
     @Override
     public <T> void outputWindowedValue(TupleTag<T> tag, WindowedValue<T> windowedValue) {
+      checkTimestamp(elem.getTimestamp(), windowedValue.getTimestamp());
       SimpleDoFnRunner.this.outputWindowedValue(tag, windowedValue);
     }
 
@@ -1039,11 +1041,13 @@ public <T> void outputWindowedValue(
 
     @Override
     public void outputWindowedValue(WindowedValue<OutputT> windowedValue) {
+      checkTimestamp(timestamp(), windowedValue.getTimestamp());
       SimpleDoFnRunner.this.outputWindowedValue(mainOutputTag, windowedValue);
     }
 
     @Override
     public <T> void outputWindowedValue(TupleTag<T> tag, WindowedValue<T> windowedValue) {
+      checkTimestamp(timestamp(), windowedValue.getTimestamp());
       SimpleDoFnRunner.this.outputWindowedValue(tag, windowedValue);
     }
 
@@ -1308,11 +1312,13 @@ public <T> void outputWindowedValue(
 
     @Override
     public void outputWindowedValue(WindowedValue<OutputT> windowedValue) {
+      checkTimestamp(this.timestamp, windowedValue.getTimestamp());
       SimpleDoFnRunner.this.outputWindowedValue(mainOutputTag, windowedValue);
     }
 
     @Override
     public <T> void outputWindowedValue(TupleTag<T> tag, WindowedValue<T> windowedValue) {
+      checkTimestamp(this.timestamp, windowedValue.getTimestamp());
       SimpleDoFnRunner.this.outputWindowedValue(tag, windowedValue);
     }
 

sdks/java/core/src/main/java/org/apache/beam/sdk/values/WindowedValues.java

@@ -272,7 +272,7 @@ public static <T> WindowedValue<T> of(
       CausedByDrain causedByDrain) {
     checkArgument(paneInfo != null, "WindowedValue requires PaneInfo, but it was null");
     checkArgument(windows.size() > 0, "WindowedValue requires windows, but there were none");
-
+    checkArgument(causedByDrain != null, "WindowedValue requires CausedByDrain, but it was null");
     if (windows.size() == 1) {
       return of(
           value,
@@ -322,6 +322,7 @@ public static <T> WindowedValue<T> of(
       @Nullable Long currentRecordOffset,
       CausedByDrain causedByDrain) {
     checkArgument(paneInfo != null, "WindowedValue requires PaneInfo, but it was null");
+    checkArgument(causedByDrain != null, "WindowedValue requires CausedByDrain, but it was null");
 
     boolean isGlobal = GlobalWindow.INSTANCE.equals(window);
     if (isGlobal && BoundedWindow.TIMESTAMP_MIN_VALUE.equals(timestamp)) {