docs: private Observable<R> executeCommandAndObserve(final AbstractCommand<R> _cmd) {
          final HystrixRequestContext currentRequestContext = HystrixRequestContext.getContextForCurrentThread();
  
          final Action1<R> markEmits = new Action1<R>() {
              @Override
              public void call(R r) {
                  if (shouldOutputOnNextEvents()) {
                      executionResult = executionResult.addEvent(HystrixEventType.EMIT);
                      eventNotifier.markEvent(HystrixEventType.EMIT, commandKey);
                  }
                  if (commandIsScalar()) {
                      long latency = System.currentTimeMillis() - executionResult.getStartTimestamp();
                      eventNotifier.markEvent(HystrixEventType.SUCCESS, commandKey);
                      executionResult = executionResult.addEvent((int) latency, HystrixEventType.SUCCESS);
                      eventNotifier.markCommandExecution(getCommandKey(), properties.executionIsolationStrategy().get(), (int) latency, executionResult.getOrderedList());
                      circuitBreaker.markSuccess();
                  }
              }
          };
  
          final Action0 markOnCompleted = new Action0() {
              @Override
              public void call() {
                  if (!commandIsScalar()) {
                      long latency = System.currentTimeMillis() - executionResult.getStartTimestamp();
                      eventNotifier.markEvent(HystrixEventType.SUCCESS, commandKey);
                      executionResult = executionResult.addEvent((int) latency, HystrixEventType.SUCCESS);
                      eventNotifier.markCommandExecution(getCommandKey(), properties.executionIsolationStrategy().get(), (int) latency, executionResult.getOrderedList());
                      circuitBreaker.markSuccess();
                  }
              }
          };
  
          final Func1<Throwable, Observable<R>> handleFallback = new Func1<Throwable, Observable<R>>() {
              @Override
              public Observable<R> call(Throwable t) {
                  circuitBreaker.markNonSuccess();
                  Exception e = getExceptionFromThrowable(t);
                  executionResult = executionResult.setExecutionException(e);
                  if (e instanceof RejectedExecutionException) {
                      return handleThreadPoolRejectionViaFallback(e);
                  } else if (t instanceof HystrixTimeoutException) {
                      return handleTimeoutViaFallback();
                  } else if (t instanceof HystrixBadRequestException) {
                      return handleBadRequestByEmittingError(e);
                  } else {
                      /*
                       * Treat HystrixBadRequestException from ExecutionHook like a plain HystrixBadRequestException.
                       */
                      if (e instanceof HystrixBadRequestException) {
                          eventNotifier.markEvent(HystrixEventType.BAD_REQUEST, commandKey);
                          return Observable.error(e);
                      }
  
                      return handleFailureViaFallback(e);
                  }
              }
          };
  
          final Action1<Notification<? super R>> setRequestContext = new Action1<Notification<? super R>>() {
              @Override
              public void call(Notification<? super R> rNotification) {
                  setRequestContextIfNeeded(currentRequestContext);
              }
          };
  
          Observable<R> execution;
          if (properties.executionTimeoutEnabled().get()) {
              execution = executeCommandWithSpecifiedIsolation(_cmd)
                      .lift(new HystrixObservableTimeoutOperator<R>(_cmd));
          } else {
              execution = executeCommandWithSpecifiedIsolation(_cmd);
          }
  
          return execution.doOnNext(markEmits)
                  .doOnCompleted(markOnCompleted)
                  .onErrorResumeNext(handleFallback)
                  .doOnEach(setRequestContext);
      }