@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 testOnTimerNoSkew() {
    TimerOutputSkewingDoFn fn = new TimerOutputSkewingDoFn(Duration.ZERO, Duration.millis(5));
    DoFnRunner<KV<String, Duration>, Duration> runner =
        new SimpleDoFnRunner<>(
            null,
            fn,
            NullSideInputReader.empty(),
            null,
            null,
            Collections.emptyList(),
            mockStepContext,
            null,
            Collections.emptyMap(),
            WindowingStrategy.of(new GlobalWindows()),
            DoFnSchemaInformation.create(),
            Collections.emptyMap());

    Exception exception =
        assertThrows(
            UserCodeException.class,
            () -> {
              runner.onTimer(
                  TimerDeclaration.PREFIX + TimerOutputSkewingDoFn.TIMER_ID,
                  "",
                  null,
                  GlobalWindow.INSTANCE,
                  new Instant(0),
                  new Instant(0),
                  TimeDomain.EVENT_TIME);
            });

    assertThat(exception.getCause(), isA(IllegalArgumentException.class));
    assertThat(
        exception.getMessage(),
        allOf(
            containsString("must be no earlier"),
            containsString(
                String.format(
                    "timestamp of the current input or timer (%s)", new Instant(0).toString())),
            containsString(
                String.format(
                    "the allowed skew (%s)",
                    PeriodFormat.getDefault().print(Duration.ZERO.toPeriod())))));
  }