Timeout does not propagate request context

[vadims]

Here's an example that reproduces the issue:

public class Test {

  public static class DummyCommand extends HystrixCommand<Void> {

    public DummyCommand() {
      super(HystrixCommandGroupKey.Factory.asKey("Dummy"));
    }

    @Override
    protected Void run() throws Exception {
      Thread.sleep(2000);
      return null;
    }
  }

  public static void main(String[] args) {
    HystrixRequestContext context = HystrixRequestContext.initializeContext();

    new DummyCommand().toObservable()
        .doOnError(new Action1<Throwable>() {
          @Override
          public void call(Throwable throwable) {
            System.out.println("initialized = " + HystrixRequestContext.isCurrentThreadInitialized());
          }
        })
        .materialize()
        .toBlockingObservable().single();

    context.shutdown();
    Hystrix.reset();
  }
}

Expected: initialized = true
Actual: initialized = false

[vadims]

One way to fix this would be to snapshot the parent thread state in HystrixContextScheduler instead of HystrixContextFunc2, since a new instance gets created every time in HystrixCommand.toObservable().

Fixing it in TimeoutObservable seemed messy.

For now I'm using the following workaround:

  public static <T> Observable<T> fromHystrix(HystrixCommand<T> command) {
    final HystrixRequestContext context = HystrixRequestContext.getContextForCurrentThread();
    return command.toObservable()
        .doOnEach(new Action1<Notification<? super T>>() {
          @Override
          public void call(Notification<? super T> notification) {
            HystrixRequestContext.setContextOnCurrentThread(context);
          }
        });
  }

[benjchristensen]

Thank you for providing sample code with it. I'll take a look.

[benjchristensen]

Can you try this branch and see if it solves the problem?

Branch => https://github.com/benjchristensen/Hystrix/tree/issue-212 in pull request #213

[vadims]

The branch fixes the issue described above.

However, today we use an additional RxJava scheduler (wrapper) to copy the MDC state across threads. I'm assuming if we changed it to use com.netflix.hystrix.strategy.concurrency.HystrixConcurrencyStrategy#wrapCallable as described in #92 then the timeout observable will lose the MDC context since it doesn't call wrapCallable?

[benjchristensen]

It should apply your plugin as HystrixConcurrencyStrategy #wrapCallable is called inside HystrixContextFunc2 which is called from within HystrixContextScheduler which is wrapped around the Scheduler being executed on.

HystrixContextScheduler.java => https://github.com/Netflix/Hystrix/blob/master/hystrix-core/src/main/java/com/netflix/hystrix/strategy/concurrency/HystrixContextScheduler.java#L47

HystrixContextFunc2 => https://github.com/Netflix/Hystrix/blob/master/hystrix-core/src/main/java/com/netflix/hystrix/strategy/concurrency/HystrixContextFunc2.java#L58

[vadims]

I'm probably doing this wrong, but here's the example that I was thinking of:

public class Test {

  private static final ThreadLocal<String> requestIdThreadLocal = new ThreadLocal<>();

  public static class DummyCommand extends HystrixCommand<Void> {

    public DummyCommand() {
      super(HystrixCommandGroupKey.Factory.asKey("Dummy"));
    }

    @Override
    protected Void run() throws Exception {
      System.out.println("requestId (run) = " + requestIdThreadLocal.get());
      Thread.sleep(2000);
      return null;
    }
  }

  public static void main(String[] args) {
    HystrixPlugins.getInstance().registerConcurrencyStrategy(new HystrixConcurrencyStrategy() {
      @Override
      public <T> Callable<T> wrapCallable(final Callable<T> callable) {
        return new RequestIdCallable<>(callable);
      }
    });

    HystrixRequestContext context = HystrixRequestContext.initializeContext();

    requestIdThreadLocal.set("foobar");

    new DummyCommand().toObservable()
        .doOnError(new Action1<Throwable>() {
          @Override
          public void call(Throwable throwable) {
            System.out.println("initialized = " + HystrixRequestContext.isCurrentThreadInitialized());
            System.out.println("requestId (timeout) = " + requestIdThreadLocal.get());
          }
        })
        .materialize()
        .toBlockingObservable().single();

    context.shutdown();
    Hystrix.reset();
  }

  private static class RequestIdCallable<T> implements Callable<T> {
    private final Callable<T> callable;
    private final String requestId;

    public RequestIdCallable(Callable<T> callable) {
      this.callable = callable;
      this.requestId = requestIdThreadLocal.get();
    }

    @Override
    public T call() throws Exception {
      String original = requestIdThreadLocal.get();
      requestIdThreadLocal.set(requestId);
      try {
        return callable.call();
      } finally {
        requestIdThreadLocal.set(original);
      }
    }
  }
}

The output I see (using the issue-212 branch) is:

requestId (run) = foobar
initialized = true
requestId (timeout) = null

however, I expected to see:

requestId (run) = foobar
initialized = true
requestId (timeout) = foobar

[benjchristensen]

That is definitely a bug. Fixing now.

[benjchristensen]

Can you confirm whether this has been fixed for you? I released it in 1.3.10 and it should also be fixed on the master branch (what will become 1.4).

[vadims]

Thanks, the issue has been fixed!

[benjchristensen]

Excellent, thanks for the confirmation.