StreamingModeExecutionContext extends DataflowExecutionContext<StepContext> { public void start( @Nullable Object key, Windmill.WorkItem work, Instant inputDataWatermark, @Nullable Instant outputDataWatermark, @Nullable Instant synchronizedProcessingTime, WindmillStateReader stateReader, StateFetcher stateFetcher, Windmill.WorkItemCommitRequest.Builder outputBuilder) { this.key = key; this.work = work; this.computationKey = WindmillComputationKey.create(computationId, work.getKey(), work.getShardingKey()); this.stateFetcher = stateFetcher; this.outputBuilder = outputBuilder; this.sideInputCache.clear(); clearSinkFullHint(); Instant processingTime = Instant.now(); for (Windmill.Timer timer : work.getTimers().getTimersList()) { if (timer.getType() == Windmill.Timer.Type.REALTIME) { Instant inferredFiringTime = WindmillTimeUtils.windmillToHarnessTimestamp(timer.getTimestamp()) .plus(Duration.millis(1)); if (inferredFiringTime.isAfter(processingTime)) { processingTime = inferredFiringTime; } } } for (StepContext stepContext : getAllStepContexts()) { stepContext.start( stateReader, inputDataWatermark, processingTime, outputDataWatermark, synchronizedProcessingTime); } } StreamingModeExecutionContext(
      CounterFactory counterFactory,
      String computationId,
      ReaderCache readerCache,
      Map<String, String> stateNameMap,
      WindmillStateCache.ForComputation stateCache,
      MetricsContainerRegistry<StreamingStepMetricsContainer> metricsContainerRegistry,
      DataflowExecutionStateTracker executionStateTracker,
      StreamingModeExecutionStateRegistry executionStateRegistry,
      long sinkByteLimit); @VisibleForTesting long getBacklogBytes(); void start(
      @Nullable Object key,
      Windmill.WorkItem work,
      Instant inputDataWatermark,
      @Nullable Instant outputDataWatermark,
      @Nullable Instant synchronizedProcessingTime,
      WindmillStateReader stateReader,
      StateFetcher stateFetcher,
      Windmill.WorkItemCommitRequest.Builder outputBuilder); @Override StepContext createStepContext(DataflowOperationContext operationContext); Iterable<Windmill.GlobalDataId> getSideInputNotifications(); @Nullable ByteString getSerializedKey(); WindmillComputationKey getComputationKey(); long getWorkToken(); Windmill.WorkItem getWork(); Windmill.WorkItemCommitRequest.Builder getOutputBuilder(); UnboundedSource.UnboundedReader<?> getCachedReader(); void setActiveReader(UnboundedSource.UnboundedReader<?> reader); void invalidateCache(); UnboundedSource.CheckpointMark getReaderCheckpoint(
      Coder<? extends UnboundedSource.CheckpointMark> coder); Map<Long, Runnable> flushState();  }
@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(); Instant now = Instant.now(); long offsetMillis = 60 * 1000; Instant timerTimestamp = now.plus(offsetMillis); timerBuilder .setTag(ByteString.copyFromUtf8("a")) .setTimestamp(timerTimestamp.getMillis() * 1000) .setType(Windmill.Timer.Type.REALTIME); executionContext.start( "key", workItemBuilder.build(), new Instant(1000), null, null, stateReader, stateFetcher, outputBuilder); TimerInternals timerInternals = stepContext.timerInternals(); assertTrue(timerTimestamp.isBefore(timerInternals.currentProcessingTime())); }
@Test(timeout = 2000) public void stateSamplingInStreaming() { StreamingModeExecutionState state = new StreamingModeExecutionState( NameContextsForTests.nameContextForTest(), "testState", null, NoopProfileScope.NOOP, null); 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); 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(); } }