StreamingModeExecutionContext extends DataflowExecutionContext<StepContext> { @Override protected SideInputReader getSideInputReaderForViews( Iterable<? extends PCollectionView<?>> views) { return StreamingModeSideInputReader.of(views, this); } 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 testSideInputReaderReconstituted() { Pipeline p = Pipeline.create(); PCollectionView<String> preview1 = p.apply(Create.of("")).apply(View.asSingleton()); PCollectionView<String> preview2 = p.apply(Create.of("")).apply(View.asSingleton()); PCollectionView<String> preview3 = p.apply(Create.of("")).apply(View.asSingleton()); SideInputReader sideInputReader = executionContext.getSideInputReaderForViews(Arrays.asList(preview1, preview2)); assertTrue(sideInputReader.contains(preview1)); assertTrue(sideInputReader.contains(preview2)); assertFalse(sideInputReader.contains(preview3)); PCollectionView<String> view1 = SerializableUtils.ensureSerializable(preview1); PCollectionView<String> view2 = SerializableUtils.ensureSerializable(preview2); PCollectionView<String> view3 = SerializableUtils.ensureSerializable(preview3); assertTrue(sideInputReader.contains(view1)); assertTrue(sideInputReader.contains(view2)); assertFalse(sideInputReader.contains(view3)); }