OutputT apply(InputT input) throws UserCodeExecutionException {
    Optional<UserCodeExecutionException> latestError = Optional.empty();
    long waitFor = 0L;
    while (waitFor != BackOff.STOP) {
      try {
        sleepIfNeeded(waitFor);
        incIfPresent(getCallCounter());
        return getThrowableFunction().apply(input);
      } catch (UserCodeExecutionException e) {
        if (!e.shouldRepeat()) {
          throw e;
        }
        latestError = Optional.of(e);
      } catch (InterruptedException ignored) {
      }
      try {
        incIfPresent(getBackoffCounter());
        waitFor = getBackOff().nextBackOffMillis();
      } catch (IOException e) {
        throw new UserCodeExecutionException(e);
      }
    }
    throw latestError.orElse(
        new UserCodeExecutionException("failed to process for input: " + input));
  }
@Test
  public void givenCallerRemoteSystemExceptionExceedsLimit_emitsIntoFailurePCollection() {
    PCollectionTuple pct =
        pipeline
            .apply(Create.of(1))
            .apply(
                ParDo.of(
                        new DoFnWithRepeaters(
                            new CallerImpl(LIMIT + 1, UserCodeRemoteSystemException.class),
                            new SetupTeardownImpl(0)))
                    .withOutputTags(OUTPUT_TAG, TupleTagList.of(FAILURE_TAG)));

    PAssert.that(pct.get(OUTPUT_TAG)).empty();
    PAssert.that(pct.get(FAILURE_TAG))
        .containsInAnyOrder(UserCodeRemoteSystemException.class.getName());

    pipeline.run();
  }