subprojects/parseq/src/main/java/com/linkedin/parseq/FusionTask.java
private Consumer3<FusionTraceContext, Promise<S>, Settable<T>> completing(
      final Consumer3<FusionTraceContext, Promise<S>, Settable<T>> propagator) {
    return (traceContext, src, dest) -> {
      final SettablePromise<T> settable = FusionTask.this.getSettableDelegate();

      if (isPropagationInitiator(traceContext)) {
        /* BaseTask's code handles completing the parent task
         * we need to handle tracing of a surrogate task here
         */
        propagator.accept(traceContext, src, new Settable<T>() {
          @Override
          public void done(T value) throws PromiseResolvedException {
            final ShallowTraceBuilder shallowTraceBuilder = traceContext.getSurrogate();
            if (shallowTraceBuilder != null) {
              addRelationships(traceContext);
              final long endNanos = System.nanoTime();
              shallowTraceBuilder.setPendingNanos(endNanos);
              shallowTraceBuilder.setEndNanos(endNanos);
              final Function<T, String> traceValueProvider = _traceValueProvider;
              shallowTraceBuilder.setResultType(ResultType.SUCCESS);
              if (traceValueProvider != null) {
                try {
                  shallowTraceBuilder.setValue(traceValueProvider.apply(value));
                } catch (Exception e) {
                  shallowTraceBuilder.setValue(Exceptions.failureToString(e));
                }
              }
            }
            dest.done(value);
          }

          @Override
          public void fail(Throwable error) throws PromiseResolvedException {
            final ShallowTraceBuilder shallowTraceBuilder = traceContext.getSurrogate();
            if (shallowTraceBuilder != null) {
              addRelationships(traceContext);
              final long endNanos = System.nanoTime();
              shallowTraceBuilder.setPendingNanos(endNanos);
              shallowTraceBuilder.setEndNanos(endNanos);
              if (Exceptions.isEarlyFinish(error)) {
                shallowTraceBuilder.setResultType(ResultType.EARLY_FINISH);
              } else {
                shallowTraceBuilder.setResultType(ResultType.ERROR);
                shallowTraceBuilder.setValue(Exceptions.failureToString(error));
              }
            }
            dest.fail(error);
          }
        });
      } else if (transitionRun(traceContext.getParent().getTraceBuilder())) {
        markTaskStarted();
        //non-parent task executed for the first time
        traceContext.getParent().getTaskLogger().logTaskStart(this);

        Settable<T> next = new Settable<T>() {
          @Override
          public void done(final T value) throws PromiseResolvedException {
            try {
              transitionDone(traceContext);
              final Function<T, String> traceValueProvider = _traceValueProvider;
              _shallowTraceBuilder.setResultType(ResultType.SUCCESS);
              if (traceValueProvider != null) {
                try {
                  _shallowTraceBuilder.setValue(traceValueProvider.apply(value));
                } catch (Exception e) {
                  _shallowTraceBuilder.setValue(Exceptions.failureToString(e));
                }
              }
              settable.done(value);
              traceContext.getParent().getTaskLogger().logTaskEnd(FusionTask.this, _traceValueProvider);
              CONTINUATIONS.submit(() -> dest.done(value));
            } catch (Exception e) {
              CONTINUATIONS.submit(() -> dest.fail(e));
            }
          }

          @Override
          public void fail(final Throwable error) throws PromiseResolvedException {
            try {
              transitionDone(traceContext);
              traceFailure(error);
              settable.fail(error);
              traceContext.getParent().getTaskLogger().logTaskEnd(FusionTask.this, _traceValueProvider);
              CONTINUATIONS.submit(() -> dest.fail(error));
            } catch (Exception e) {
              CONTINUATIONS.submit(() -> dest.fail(e));
            }
          }
        };
        CONTINUATIONS.submit(() -> {
          try {
            propagator.accept(traceContext, src, next);
          } catch (Exception e) {
            /* This can only happen if there is an internal problem. Propagators should not throw any exceptions. */
            LOGGER.error("ParSeq ingternal error. An exception was thrown by propagator.", e);
          }
        });
      } else {
        //non-parent tasks subsequent executions
        addPotentialRelationships(traceContext, traceContext.getParent().getTraceBuilder());
        Promises.propagateResult(settable, dest);
      }
    };
  }
private Consumer3<FusionTraceContext, Promise<S>, Settable<T>> completing(
      final Consumer3<FusionTraceContext, Promise<S>, Settable<T>> propagator) {
    return (traceContext, src, dest) -> {
      final SettablePromise<T> settable = FusionTask.this.getSettableDelegate();

      if (isPropagationInitiator(traceContext)) {
        /* BaseTask's code handles completing the parent task
         * we need to handle tracing of a surrogate task here
         */
        propagator.accept(traceContext, src, new Settable<T>() {
          @Override
          public void done(T value) throws PromiseResolvedException {
            final ShallowTraceBuilder shallowTraceBuilder = traceContext.getSurrogate();
            if (shallowTraceBuilder != null) {
              addRelationships(traceContext);
              final long endNanos = System.nanoTime();
              shallowTraceBuilder.setPendingNanos(endNanos);
              shallowTraceBuilder.setEndNanos(endNanos);
              final Function<T, String> traceValueProvider = _traceValueProvider;
              shallowTraceBuilder.setResultType(ResultType.SUCCESS);
              if (traceValueProvider != null) {
                try {
                  shallowTraceBuilder.setValue(traceValueProvider.apply(value));
                } catch (Exception e) {
                  shallowTraceBuilder.setValue(Exceptions.failureToString(e));
                }
              }
            }
            dest.done(value);
          }

          @Override
          public void fail(Throwable error) throws PromiseResolvedException {
            final ShallowTraceBuilder shallowTraceBuilder = traceContext.getSurrogate();
            if (shallowTraceBuilder != null) {
              addRelationships(traceContext);
              final long endNanos = System.nanoTime();
              shallowTraceBuilder.setPendingNanos(endNanos);
              shallowTraceBuilder.setEndNanos(endNanos);
              if (Exceptions.isEarlyFinish(error)) {
                shallowTraceBuilder.setResultType(ResultType.EARLY_FINISH);
              } else {
                shallowTraceBuilder.setResultType(ResultType.ERROR);
                shallowTraceBuilder.setValue(Exceptions.failureToString(error));
              }
            }
            dest.fail(error);
          }
        });
      } else if (transitionRun(traceContext.getParent().getTraceBuilder())) {
        markTaskStarted();
        //non-parent task executed for the first time
        traceContext.getParent().getTaskLogger().logTaskStart(this);

        Settable<T> next = new Settable<T>() {
          @Override
          public void done(final T value) throws PromiseResolvedException {
            try {
              transitionDone(traceContext);
              final Function<T, String> traceValueProvider = _traceValueProvider;
              _shallowTraceBuilder.setResultType(ResultType.SUCCESS);
              if (traceValueProvider != null) {
                try {
                  _shallowTraceBuilder.setValue(traceValueProvider.apply(value));
                } catch (Exception e) {
                  _shallowTraceBuilder.setValue(Exceptions.failureToString(e));
                }
              }
              settable.done(value);
              traceContext.getParent().getTaskLogger().logTaskEnd(FusionTask.this, _traceValueProvider);
              CONTINUATIONS.submit(() -> dest.done(value));
            } catch (Exception e) {
              CONTINUATIONS.submit(() -> dest.fail(e));
            }
          }

          @Override
          public void fail(final Throwable error) throws PromiseResolvedException {
            try {
              transitionDone(traceContext);
              traceFailure(error);
              settable.fail(error);
              traceContext.getParent().getTaskLogger().logTaskEnd(FusionTask.this, _traceValueProvider);
              CONTINUATIONS.submit(() -> dest.fail(error));
            } catch (Exception e) {
              CONTINUATIONS.submit(() -> dest.fail(e));
            }
          }
        };
        CONTINUATIONS.submit(() -> {
          try {
            propagator.accept(traceContext, src, next);
          } catch (Exception e) {
            /* This can only happen if there is an internal problem. Propagators should not throw any exceptions. */
            LOGGER.error("ParSeq ingternal error. An exception was thrown by propagator.", e);
          }
        });
      } else {
        //non-parent tasks subsequent executions
        addPotentialRelationships(traceContext, traceContext.getParent().getTraceBuilder());
        Promises.propagateResult(settable, dest);
      }
    };
  }
Adds closure to the propagator that maintains correct state of the task e.g. transitions between states,
adds trace etc.