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));
  }
@Test
  public void getExecutionContextDifferentStepsIndependentState() {
    StructuralKey<?> myKey = StructuralKey.of("foo", StringUtf8Coder.of());
    DirectExecutionContext fooContext = context.getExecutionContext(impulseProducer, 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());
  }