SimpleDoFnRunner implements DoFnRunner<InputT, OutputT> { @Override public <KeyT> void onTimer( String timerId, String timerFamilyId, KeyT key, BoundedWindow window, Instant timestamp, Instant outputTimestamp, TimeDomain timeDomain) { Instant effectiveTimestamp; switch (timeDomain) { case EVENT_TIME: effectiveTimestamp = outputTimestamp; break; case PROCESSING_TIME: case SYNCHRONIZED_PROCESSING_TIME: effectiveTimestamp = stepContext.timerInternals().currentInputWatermarkTime(); break; default: throw new IllegalArgumentException(String.format("Unknown time domain: %s", timeDomain)); } OnTimerArgumentProvider<KeyT> argumentProvider = new OnTimerArgumentProvider<>( timerId, key, window, timestamp, effectiveTimestamp, timeDomain); invoker.invokeOnTimer(timerId, timerFamilyId, argumentProvider); } 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 testOnTimerExceptionsWrappedAsUserCodeException() { 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.onTimer( TimerDeclaration.PREFIX + ThrowingDoFn.TIMER_ID, "", null, GlobalWindow.INSTANCE, new Instant(0), new Instant(0), TimeDomain.EVENT_TIME); }
@Test public void testOnTimerCalled() { WindowFn<?, GlobalWindow> 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(windowFn), DoFnSchemaInformation.create(), Collections.emptyMap()); Instant currentTime = new Instant(42); Duration offset = Duration.millis(37); runner.onTimer( TimerDeclaration.PREFIX + DoFnWithTimers.TIMER_ID, "", null, GlobalWindow.INSTANCE, currentTime.plus(offset), currentTime.plus(offset), TimeDomain.EVENT_TIME); assertThat( fn.onTimerInvocations, contains( TimerData.of( DoFnWithTimers.TIMER_ID, "", StateNamespaces.window(windowFn.windowCoder(), GlobalWindow.INSTANCE), currentTime.plus(offset), currentTime.plus(offset), TimeDomain.EVENT_TIME))); }