public void start(
      @Nullable Object key,
      Work work,
      WindmillStateReader stateReader,
      SideInputStateFetcher sideInputStateFetcher,
      OperationalLimits operationalLimits,
      Windmill.WorkItemCommitRequest.Builder outputBuilder) {
    this.key = key;
    this.work = work;
    this.computationKey = WindmillComputationKey.create(computationId, work.getShardedKey());
    this.sideInputStateFetcher = sideInputStateFetcher;
    this.operationalLimits = operationalLimits;
    this.outputBuilder = outputBuilder;
    this.sideInputCache.clear();
    clearSinkFullHint();

    Instant processingTime = computeProcessingTime(work.getWorkItem().getTimers().getTimersList());

    Collection<? extends StepContext> stepContexts = getAllStepContexts();
    if (!stepContexts.isEmpty()) {
      // This must be only created once for the workItem as token validation will fail if the same
      // work token is reused.
      WindmillStateCache.ForKey cacheForKey =
          stateCache.forKey(getComputationKey(), getWorkItem().getCacheToken(), getWorkToken());
      for (StepContext stepContext : stepContexts) {
        stepContext.start(stateReader, processingTime, cacheForKey, work.watermarks());
      }
    }
  }
@Test
  public void testTimerInternalsProcessingTimeSkew() {
    Windmill.WorkItemCommitRequest.Builder outputBuilder =
        Windmill.WorkItemCommitRequest.newBuilder();

    NameContext nameContext = NameContextsForTests.nameContextForTest();
    DataflowOperationContext operationContext =
        executionContext.createOperationContext(nameContext);
    StreamingModeExecutionContext.StepContext stepContext =
        executionContext.getStepContext(operationContext);
    Windmill.WorkItem.Builder workItemBuilder =
        Windmill.WorkItem.newBuilder().setKey(ByteString.EMPTY).setWorkToken(17L);
    Windmill.Timer.Builder timerBuilder = workItemBuilder.getTimersBuilder().addTimersBuilder();

    // Trigger a realtime timer that with clock skew but ensure that it would
    // still fire.
    Instant now = Instant.now();
    long offsetMillis = 60 * 1000;
    Instant timerTimestamp = now.plus(Duration.millis(offsetMillis));
    timerBuilder
        .setTag(ByteString.copyFromUtf8("a"))
        .setTimestamp(timerTimestamp.getMillis() * 1000)
        .setType(Windmill.Timer.Type.REALTIME);

    executionContext.start(
        "key",
        createMockWork(
            workItemBuilder.build(),
            Watermarks.builder().setInputDataWatermark(new Instant(1000)).build()),
        stateReader,
        sideInputStateFetcher,
        OperationalLimits.builder().build(),
        outputBuilder);
    TimerInternals timerInternals = stepContext.timerInternals();
    assertTrue(timerTimestamp.isBefore(timerInternals.currentProcessingTime()));
  }