subprojects/parseq/src/main/java/com/linkedin/parseq/retry/RetriableTask.java
private Task<T> wrap(int attempt) {
    Task<T> retryTask = Task.async(_policy.getName() + ", attempt " + attempt, context -> {
      final SettablePromise<T> result = Promises.settable();

      Task<T> task = _taskFunction.apply(attempt);

      final Task<T> recovery = Task.async(_name + " recovery", recoveryContext -> {
        final SettablePromise<T> recoveryResult = Promises.settable();

        if (task.isFailed()) {
          // Failed task will cause retry to be scheduled.
          ErrorClassification errorClassification = _policy.getErrorClassifier().apply(task.getError());
          retry(attempt + 1, task.getError(), errorClassification, recoveryContext, recoveryResult);
        } else {
          recoveryResult.done(task.get());
        }

        return recoveryResult;
      });

      // Recovery task should run immediately after the original task to process its error.
      recovery.setPriority(Priority.MAX_PRIORITY);
      recovery.getShallowTraceBuilder().setSystemHidden(true);
      Promises.propagateResult(recovery, result);
      context.after(task).run(recovery);
      context.run(task);

      return result;
    });
    retryTask.getShallowTraceBuilder().setTaskType(TaskType.RETRY.getName());
    return retryTask;
  }
private Task<T> wrap(int attempt) {
    Task<T> retryTask = Task.async(_policy.getName() + ", attempt " + attempt, context -> {
      final SettablePromise<T> result = Promises.settable();

      Task<T> task = _taskFunction.apply(attempt);

      final Task<T> recovery = Task.async(_name + " recovery", recoveryContext -> {
        final SettablePromise<T> recoveryResult = Promises.settable();

        if (task.isFailed()) {
          // Failed task will cause retry to be scheduled.
          ErrorClassification errorClassification = _policy.getErrorClassifier().apply(task.getError());
          retry(attempt + 1, task.getError(), errorClassification, recoveryContext, recoveryResult);
        } else {
          recoveryResult.done(task.get());
        }

        return recoveryResult;
      });

      // Recovery task should run immediately after the original task to process its error.
      recovery.setPriority(Priority.MAX_PRIORITY);
      recovery.getShallowTraceBuilder().setSystemHidden(true);
      Promises.propagateResult(recovery, result);
      context.after(task).run(recovery);
      context.run(task);

      return result;
    });
    retryTask.getShallowTraceBuilder().setTaskType(TaskType.RETRY.getName());
    return retryTask;
  }
Create a wrapped task with associated recovery task that will retry if necessary.