SimpleDoFnRunner implements DoFnRunner<InputT, OutputT> { @Override public void processElement(WindowedValue<InputT> compressedElem) { if (observesWindow) { for (WindowedValue<InputT> elem : compressedElem.explodeWindows()) { invokeProcessElement(elem); } } else { invokeProcessElement(compressedElem); } } SimpleDoFnRunner(
      PipelineOptions options,
      DoFn<InputT, OutputT> fn,
      SideInputReader sideInputReader,
      OutputManager outputManager,
      TupleTag<OutputT> mainOutputTag,
      List<TupleTag<?>> additionalOutputTags,
      StepContext stepContext,
      @Nullable Coder<InputT> inputCoder,
      Map<TupleTag<?>, Coder<?>> outputCoders,
      WindowingStrategy<?, ?> windowingStrategy,
      DoFnSchemaInformation doFnSchemaInformation,
      Map<String, PCollectionView<?>> sideInputMapping); @Override DoFn<InputT, OutputT> getFn(); @Override void startBundle(); @Override void processElement(WindowedValue<InputT> compressedElem); @Override void onTimer(
      String timerId,
      String timerFamilyId,
      KeyT key,
      BoundedWindow window,
      Instant timestamp,
      Instant outputTimestamp,
      TimeDomain timeDomain); @Override void finishBundle(); @Override void onWindowExpiration(BoundedWindow window, Instant timestamp, KeyT key);  }
@Test public void testProcessElementExceptionsWrappedAsUserCodeException() { ThrowingDoFn fn = new ThrowingDoFn(); DoFnRunner<String, String> runner = new SimpleDoFnRunner<>( null, fn, NullSideInputReader.empty(), null, null, Collections.emptyList(), mockStepContext, null, Collections.emptyMap(), WindowingStrategy.of(new GlobalWindows()), DoFnSchemaInformation.create(), Collections.emptyMap()); thrown.expect(UserCodeException.class); thrown.expectCause(is(fn.exceptionToThrow)); runner.processElement(WindowedValue.valueInGlobalWindow("anyValue")); }
@Test public void testTimerSet() { WindowFn<?, ?> windowFn = new GlobalWindows(); DoFnWithTimers<GlobalWindow> fn = new DoFnWithTimers(windowFn.windowCoder()); DoFnRunner<String, String> runner = new SimpleDoFnRunner<>( null, fn, NullSideInputReader.empty(), null, null, Collections.emptyList(), mockStepContext, null, Collections.emptyMap(), WindowingStrategy.of(new GlobalWindows()), DoFnSchemaInformation.create(), Collections.emptyMap()); Instant currentTime = new Instant(42); when(mockTimerInternals.currentInputWatermarkTime()).thenReturn(currentTime); runner.processElement(WindowedValue.valueInGlobalWindow("anyValue")); verify(mockTimerInternals) .setTimer( StateNamespaces.window(new GlobalWindows().windowCoder(), GlobalWindow.INSTANCE), TimerDeclaration.PREFIX + DoFnWithTimers.TIMER_ID, "", currentTime.plus(DoFnWithTimers.TIMER_OFFSET), currentTime.plus(DoFnWithTimers.TIMER_OFFSET), TimeDomain.EVENT_TIME); }