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(timeout = 2000)
  public void stateSamplingInStreaming() {
    // Test that when writing on one thread and reading from another, updates always eventually
    // reach the reading thread.
    StreamingModeExecutionState state =
        new StreamingModeExecutionState(
            NameContextsForTests.nameContextForTest(), "testState", null, NoopProfileScope.NOOP);
    ExecutionStateSampler sampler = ExecutionStateSampler.newForTest();
    try {
      sampler.start();

      ExecutionStateTracker tracker = new ExecutionStateTracker(sampler);

      Thread executionThread = new Thread();
      executionThread.setName("looping-thread-for-test");

      tracker.activate(executionThread);
      tracker.enterState(state);

      // Wait for the state to be incremented 3 times
      for (int i = 0; i < 3; i++) {
        CounterUpdate update = null;
        while (update == null) {
          update = state.extractUpdate(false);
        }
        long newValue = splitIntToLong(update.getInteger());
        assertThat(newValue, Matchers.greaterThan(0L));
      }
    } finally {
      sampler.stop();
    }
  }