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);
      }
    };
  }
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.
Adds closure to the propagator that maintains correct state of the task e . g . transitions between states adds trace etc .