This closes #1619
diff --git a/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java b/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
index 0d41a8d..af7f5ca 100644
--- a/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
+++ b/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
@@ -114,7 +114,7 @@
WindowingStrategy<?, ?> windowingStrategy) {
this.fn = fn;
this.signature = DoFnSignatures.getSignature(fn.getClass());
- this.observesWindow = signature.processElement().observesWindow();
+ this.observesWindow = signature.processElement().observesWindow() || !sideInputReader.isEmpty();
this.invoker = DoFnInvokers.invokerFor(fn);
this.outputManager = outputManager;
this.mainOutputTag = mainOutputTag;
diff --git a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactoryTest.java b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactoryTest.java
index 7c086a1..326310b 100644
--- a/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactoryTest.java
+++ b/runners/direct-java/src/test/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactoryTest.java
@@ -31,6 +31,7 @@
import com.google.common.collect.Lists;
import java.io.Serializable;
import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import org.apache.beam.runners.direct.DirectRunner.CommittedBundle;
import org.apache.beam.runners.direct.DirectRunner.UncommittedBundle;
@@ -94,6 +95,11 @@
public void setup() {
MockitoAnnotations.initMocks(this);
when((StateInternals<Object>) mockStepContext.stateInternals()).thenReturn(stateInternals);
+ when(mockEvaluationContext.createSideInputReader(anyList()))
+ .thenReturn(
+ SideInputContainer.create(
+ mockEvaluationContext, Collections.<PCollectionView<?>>emptyList())
+ .createReaderForViews(Collections.<PCollectionView<?>>emptyList()));
}
@Test
