@Override
  public Iterator<RawUnionValue> call(Iterator<WindowedValue<InputT>> inputs) throws Exception {
    SparkPipelineOptions options = pipelineOptions.get().as(SparkPipelineOptions.class);
    // Register standard file systems.
    FileSystems.setDefaultPipelineOptions(options);

    // Do not call processElements if there are no inputs
    // Otherwise, this may cause validation errors (e.g. ParDoTest)
    if (!inputs.hasNext()) {
      return Collections.emptyIterator();
    }

    try (ExecutableStageContext stageContext = contextFactory.get(jobInfo)) {
      ExecutableStage executableStage = ExecutableStage.fromPayload(stagePayload);
      try (StageBundleFactory stageBundleFactory =
          stageContext.getStageBundleFactory(executableStage)) {
        ConcurrentLinkedQueue<RawUnionValue> collector = new ConcurrentLinkedQueue<>();
        StateRequestHandler stateRequestHandler =
            getStateRequestHandler(
                executableStage, stageBundleFactory.getProcessBundleDescriptor());
        if (executableStage.getTimers().size() == 0) {
          ReceiverFactory receiverFactory = new ReceiverFactory(collector, outputMap);
          processElements(stateRequestHandler, receiverFactory, null, stageBundleFactory, inputs);
          return collector.iterator();
        }
        // Used with Batch, we know that all the data is available for this key. We can't use the
        // timer manager from the context because it doesn't exist. So we create one and advance
        // time to the end after processing all elements.
        final InMemoryTimerInternals timerInternals = new InMemoryTimerInternals();
        timerInternals.advanceProcessingTime(Instant.now());
        timerInternals.advanceSynchronizedProcessingTime(Instant.now());

        ReceiverFactory receiverFactory = new ReceiverFactory(collector, outputMap);

        TimerReceiverFactory timerReceiverFactory =
            new TimerReceiverFactory(
                stageBundleFactory,
                (Timer<?> timer, TimerInternals.TimerData timerData) -> {
                  currentTimerKey = timer.getUserKey();
                  if (timer.getClearBit()) {
                    timerInternals.deleteTimer(timerData);
                  } else {
                    timerInternals.setTimer(timerData);
                  }
                },
                windowCoder);

        // Process inputs.
        processElements(
            stateRequestHandler, receiverFactory, timerReceiverFactory, stageBundleFactory, inputs);

        // Finish any pending windows by advancing the input watermark to infinity.
        timerInternals.advanceInputWatermark(BoundedWindow.TIMESTAMP_MAX_VALUE);
        // Finally, advance the processing time to infinity to fire any timers.
        timerInternals.advanceProcessingTime(BoundedWindow.TIMESTAMP_MAX_VALUE);
        timerInternals.advanceSynchronizedProcessingTime(BoundedWindow.TIMESTAMP_MAX_VALUE);

        // Now we fire the timers and process elements generated by timers (which may be timers
        // itself)
        while (timerInternals.hasPendingTimers()) {
          try (RemoteBundle bundle =
              stageBundleFactory.getBundle(
                  receiverFactory,
                  timerReceiverFactory,
                  stateRequestHandler,
                  getBundleProgressHandler())) {

            PipelineTranslatorUtils.fireEligibleTimers(
                timerInternals, bundle.getTimerReceivers(), currentTimerKey);
          }
        }
        return collector.iterator();
      }
    }
  }
@Test
  public void expectedInputsAreSent() throws Exception {
    SparkExecutableStageFunction<Integer, ?> function = getFunction(Collections.emptyMap());

    RemoteBundle bundle = Mockito.mock(RemoteBundle.class);
    when(stageBundleFactory.getBundle(any(), any(), any(), any(BundleProgressHandler.class)))
        .thenReturn(bundle);

    @SuppressWarnings("unchecked")
    FnDataReceiver<WindowedValue<?>> receiver = Mockito.mock(FnDataReceiver.class);
    when(bundle.getInputReceivers()).thenReturn(ImmutableMap.of(inputId, receiver));

    WindowedValue<Integer> one = WindowedValue.valueInGlobalWindow(1);
    WindowedValue<Integer> two = WindowedValue.valueInGlobalWindow(2);
    WindowedValue<Integer> three = WindowedValue.valueInGlobalWindow(3);
    function.call(Arrays.asList(one, two, three).iterator());

    verify(receiver).accept(one);
    verify(receiver).accept(two);
    verify(receiver).accept(three);
    verifyNoMoreInteractions(receiver);
  }