@Override
  public <T> TransformEvaluator<T> forApplication(
      AppliedPTransform<?, ?, ?> application, CommittedBundle<?> inputBundle) throws Exception {
    @SuppressWarnings({"unchecked", "rawtypes"})
    TransformEvaluator<T> evaluator =
        (TransformEvaluator<T>)
            createEvaluator((AppliedPTransform) application, (CommittedBundle) inputBundle);
    return evaluator;
  }
@Test
  public void testUnprocessedElements() throws Exception {
    // To test the factory, first we set up a pipeline and then we use the constructed
    // pipeline to create the right parameters to pass to the factory

    final String stateId = "my-state-id";

    // For consistency, window it into FixedWindows. Actually we will fabricate an input bundle.
    PCollection<KV<String, Integer>> mainInput =
        pipeline
            .apply(Create.of(KV.of("hello", 1), KV.of("hello", 2)))
            .apply(Window.into(FixedWindows.of(Duration.millis(10))));

    final PCollectionView<List<Integer>> sideInput =
        pipeline
            .apply("Create side input", Create.of(42))
            .apply("Window side input", Window.into(FixedWindows.of(Duration.millis(10))))
            .apply("View side input", View.asList());

    TupleTag<Integer> mainOutput = new TupleTag<>();
    PCollection<Integer> produced =
        mainInput
            .apply(
                new ParDoMultiOverrideFactory.GbkThenStatefulParDo<>(
                    new DoFn<KV<String, Integer>, Integer>() {

                      @StateId(stateId)
                      private final StateSpec<ValueState<String>> spec =
                          StateSpecs.value(StringUtf8Coder.of());

                      @ProcessElement
                      public void process(ProcessContext c) {}
                    },
                    mainOutput,
                    TupleTagList.empty(),
                    Collections.singletonList(sideInput),
                    DoFnSchemaInformation.create(),
                    Collections.emptyMap()))
            .get(mainOutput)
            .setCoder(VarIntCoder.of());

    StatefulParDoEvaluatorFactory<String, Integer, Integer> factory =
        new StatefulParDoEvaluatorFactory<>(mockEvaluationContext, options);

    // This will be the stateful ParDo from the expansion
    AppliedPTransform<
            PCollection<KeyedWorkItem<String, KV<String, Integer>>>,
            PCollectionTuple,
            StatefulParDo<String, Integer, Integer>>
        producingTransform = (AppliedPTransform) DirectGraphs.getProducer(produced);

    // Then there will be a digging down to the step context to get the state internals
    when(mockEvaluationContext.getExecutionContext(
            eq(producingTransform), Mockito.<StructuralKey>any()))
        .thenReturn(mockExecutionContext);
    when(mockExecutionContext.getStepContext(any())).thenReturn(mockStepContext);
    when(mockEvaluationContext.createBundle(Matchers.<PCollection<Integer>>any()))
        .thenReturn(mockUncommittedBundle);
    when(mockStepContext.getTimerUpdate()).thenReturn(TimerUpdate.empty());

    // And digging to check whether the window is ready
    when(mockEvaluationContext.createSideInputReader(anyList())).thenReturn(mockSideInputReader);
    when(mockSideInputReader.isReady(Matchers.any(), Matchers.any())).thenReturn(false);

    IntervalWindow firstWindow = new IntervalWindow(new Instant(0), new Instant(9));

    // A single bundle with some elements in the global window; it should register cleanup for the
    // global window state merely by having the evaluator created. The cleanup logic does not
    // depend on the window.
    String key = "hello";
    WindowedValue<KV<String, Integer>> firstKv =
        WindowedValue.of(KV.of(key, 1), new Instant(3), firstWindow, PaneInfo.NO_FIRING);

    WindowedValue<KeyedWorkItem<String, KV<String, Integer>>> gbkOutputElement =
        firstKv.withValue(
            KeyedWorkItems.elementsWorkItem(
                "hello",
                ImmutableList.of(
                    firstKv,
                    firstKv.withValue(KV.of(key, 13)),
                    firstKv.withValue(KV.of(key, 15)))));

    CommittedBundle<KeyedWorkItem<String, KV<String, Integer>>> inputBundle =
        BUNDLE_FACTORY
            .createBundle(
                (PCollection<KeyedWorkItem<String, KV<String, Integer>>>)
                    Iterables.getOnlyElement(
                        TransformInputs.nonAdditionalInputs(producingTransform)))
            .add(gbkOutputElement)
            .commit(Instant.now());
    TransformEvaluator<KeyedWorkItem<String, KV<String, Integer>>> evaluator =
        factory.forApplication(producingTransform, inputBundle);

    evaluator.processElement(gbkOutputElement);

    // This should push back every element as a KV<String, Iterable<Integer>>
    // in the appropriate window. Since the keys are equal they are single-threaded
    TransformResult<KeyedWorkItem<String, KV<String, Integer>>> result = evaluator.finishBundle();

    List<Integer> pushedBackInts = new ArrayList<>();

    for (WindowedValue<? extends KeyedWorkItem<String, KV<String, Integer>>> unprocessedElement :
        result.getUnprocessedElements()) {

      assertThat(
          Iterables.getOnlyElement(unprocessedElement.getWindows()),
          equalTo((BoundedWindow) firstWindow));

      assertThat(unprocessedElement.getValue().key(), equalTo("hello"));
      for (WindowedValue<KV<String, Integer>> windowedKv :
          unprocessedElement.getValue().elementsIterable()) {
        pushedBackInts.add(windowedKv.getValue().getValue());
      }
    }
    assertThat(pushedBackInts, containsInAnyOrder(1, 13, 15));
  }