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 givenSetupRemoteSystemErrorsExceedsLimit_throws() {
    pipeline
        .apply(Create.of(1))
        .apply(
            ParDo.of(
                    new DoFnWithRepeaters(
                        new CallerImpl(0),
                        new SetupTeardownImpl(LIMIT + 1, UserCodeRemoteSystemException.class)))
                .withOutputTags(OUTPUT_TAG, TupleTagList.of(FAILURE_TAG)));

    UncheckedExecutionException thrown =
        assertThrows(UncheckedExecutionException.class, pipeline::run);
    assertThat(thrown.getCause(), allOf(notNullValue(), instanceOf(UserCodeException.class)));
    assertThat(
        thrown.getCause().getCause(),
        allOf(notNullValue(), instanceOf(UserCodeRemoteSystemException.class)));
  }