@Override
  public <KeyT> void onTimer(
      String timerId,
      String timerFamilyId,
      KeyT key,
      BoundedWindow window,
      Instant timestamp,
      Instant outputTimestamp,
      TimeDomain timeDomain) {
    Preconditions.checkNotNull(outputTimestamp, "outputTimestamp");

    OnTimerArgumentProvider<KeyT> argumentProvider =
        new OnTimerArgumentProvider<>(timerId, key, window, timestamp, outputTimestamp, timeDomain);
    invoker.invokeOnTimer(timerId, timerFamilyId, argumentProvider);
  }
@Test
  public void testOnTimerAllowedSkew() {
    TimerOutputSkewingDoFn fn = new TimerOutputSkewingDoFn(Duration.millis(10), Duration.millis(5));
    DoFnRunner<KV<String, Duration>, Duration> runner =
        new SimpleDoFnRunner<>(
            null,
            fn,
            NullSideInputReader.empty(),
            new ListOutputManager(),
            null,
            Collections.emptyList(),
            mockStepContext,
            null,
            Collections.emptyMap(),
            WindowingStrategy.of(new GlobalWindows()),
            DoFnSchemaInformation.create(),
            Collections.emptyMap());

    runner.onTimer(
        TimerDeclaration.PREFIX + TimerOutputSkewingDoFn.TIMER_ID,
        "",
        null,
        GlobalWindow.INSTANCE,
        new Instant(0),
        new Instant(0),
        TimeDomain.EVENT_TIME);
  }