EvaluationContext { public DirectExecutionContext getExecutionContext( AppliedPTransform<?, ?, ?> application, StructuralKey<?> key) { StepAndKey stepAndKey = StepAndKey.of(application, key); return new DirectExecutionContext( clock, key, (CopyOnAccessInMemoryStateInternals) applicationStateInternals.get(stepAndKey), watermarkManager.getWatermarks(application)); } private  EvaluationContext(
      Clock clock,
      BundleFactory bundleFactory,
      DirectGraph graph,
      Set<PValue> keyedPValues,
      ExecutorService executorService); static EvaluationContext create(
      Clock clock,
      BundleFactory bundleFactory,
      DirectGraph graph,
      Set<PValue> keyedPValues,
      ExecutorService executorService); void initialize(
      Map<AppliedPTransform<?, ?, ?>, ? extends Iterable<CommittedBundle<?>>> initialInputs); CommittedResult<AppliedPTransform<?, ?, ?>> handleResult(
      CommittedBundle<?> completedBundle,
      Iterable<TimerData> completedTimers,
      TransformResult<?> result); UncommittedBundle<T> createRootBundle(); UncommittedBundle<T> createBundle(PCollection<T> output); UncommittedBundle<T> createKeyedBundle(
      StructuralKey<K> key, PCollection<T> output); boolean isKeyed(PValue pValue); PCollectionViewWriter<ElemT, ViewT> createPCollectionViewWriter(
      PCollection<Iterable<ElemT>> input, final PCollectionView<ViewT> output); void scheduleAfterOutputWouldBeProduced(
      PCollection<?> value,
      BoundedWindow window,
      WindowingStrategy<?, ?> windowingStrategy,
      Runnable runnable); void scheduleAfterOutputWouldBeProduced(
      PCollectionView<?> view,
      BoundedWindow window,
      WindowingStrategy<?, ?> windowingStrategy,
      Runnable runnable); void scheduleAfterWindowExpiration(
      AppliedPTransform<?, ?, ?> producing,
      BoundedWindow window,
      WindowingStrategy<?, ?> windowingStrategy,
      Runnable runnable); DirectExecutionContext getExecutionContext(
      AppliedPTransform<?, ?, ?> application, StructuralKey<?> key); ReadyCheckingSideInputReader createSideInputReader(
      final List<PCollectionView<?>> sideInputs); DirectMetrics getMetrics(); boolean isDone(AppliedPTransform<?, ?, ?> transform); boolean isDone(); Instant now();  }
@Test public void getExecutionContextDifferentKeysIndependentState() { DirectExecutionContext fooContext = context.getExecutionContext(createdProducer, StructuralKey.of("foo", StringUtf8Coder.of())); StateTag<BagState<Integer>> intBag = StateTags.bag("myBag", VarIntCoder.of()); fooContext.getStepContext("s1").stateInternals().state(StateNamespaces.global(), intBag).add(1); DirectExecutionContext barContext = context.getExecutionContext(createdProducer, StructuralKey.of("bar", StringUtf8Coder.of())); assertThat(barContext, not(equalTo(fooContext))); assertThat( barContext .getStepContext("s1") .stateInternals() .state(StateNamespaces.global(), intBag) .read(), emptyIterable()); }
@Test public void getExecutionContextDifferentStepsIndependentState() { StructuralKey<?> myKey = StructuralKey.of("foo", StringUtf8Coder.of()); DirectExecutionContext fooContext = context.getExecutionContext(createdProducer, myKey); StateTag<BagState<Integer>> intBag = StateTags.bag("myBag", VarIntCoder.of()); fooContext.getStepContext("s1").stateInternals().state(StateNamespaces.global(), intBag).add(1); DirectExecutionContext barContext = context.getExecutionContext(downstreamProducer, myKey); assertThat( barContext .getStepContext("s1") .stateInternals() .state(StateNamespaces.global(), intBag) .read(), emptyIterable()); }