subprojects/parseq/src/main/java/com/linkedin/parseq/FusionTask.java
private <R> Consumer3<FusionTraceContext, Promise<S>, Settable<T>> compose(
      final Consumer3<FusionTraceContext, Promise<S>, Settable<R>> predecessor,
      final Consumer3<FusionTraceContext, Promise<R>, Settable<T>> propagator) {

    return (traceContext, src, dst) -> {

      /*
       * At this point we know that transformation chain's length > 1.
       * This code is executed during task execution, not when task is constructed.
       */
      traceContext.createSurrogate();

      predecessor.accept(traceContext, src, new Settable<R>() {

        @Override
        public void done(R value) throws PromiseResolvedException {
          try {
            /* Track start time of the transformation. End time is tracked by closure created by completing() */
            getEffectiveShallowTraceBuilder(traceContext).setStartNanos(System.nanoTime());
            propagator.accept(traceContext, Promises.value(value), dst);
          } 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);
          }
        }

        @Override
        public void fail(Throwable error) throws PromiseResolvedException {
          try {
            /* Track start time of the transformation. End time is tracked by closure created by completing() */
            getEffectiveShallowTraceBuilder(traceContext).setStartNanos(System.nanoTime());
            propagator.accept(traceContext, Promises.error(error), dst);
          } 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);
          }
        }
      });
    };
  }
private <R> Consumer3<FusionTraceContext, Promise<S>, Settable<T>> compose(
      final Consumer3<FusionTraceContext, Promise<S>, Settable<R>> predecessor,
      final Consumer3<FusionTraceContext, Promise<R>, Settable<T>> propagator) {

    return (traceContext, src, dst) -> {

      /*
       * At this point we know that transformation chain's length > 1.
       * This code is executed during task execution, not when task is constructed.
       */
      traceContext.createSurrogate();

      predecessor.accept(traceContext, src, new Settable<R>() {

        @Override
        public void done(R value) throws PromiseResolvedException {
          try {
            /* Track start time of the transformation. End time is tracked by closure created by completing() */
            getEffectiveShallowTraceBuilder(traceContext).setStartNanos(System.nanoTime());
            propagator.accept(traceContext, Promises.value(value), dst);
          } 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);
          }
        }

        @Override
        public void fail(Throwable error) throws PromiseResolvedException {
          try {
            /* Track start time of the transformation. End time is tracked by closure created by completing() */
            getEffectiveShallowTraceBuilder(traceContext).setStartNanos(System.nanoTime());
            propagator.accept(traceContext, Promises.error(error), dst);
          } 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);
          }
        }
      });
    };
  }
Composes transformation with the transformation of the predecessor.