Skip to content

Commit 7891a1b

Browse files
benjchristensenbenjchristensen
committed
Merge pull request #213 from benjchristensen/issue-212
BugFix: Timeout does not propagate request context
2 parents 375b8f3 + cb9a953 commit 7891a1b

File tree

8 files changed

+187
-25
lines changed

8 files changed

+187
-25
lines changed

‎hystrix-core/src/main/java/com/netflix/hystrix/HystrixCollapser.java

+1-1
Original file line numberDiff line numberDiff line change
@@ -880,7 +880,7 @@ public void testRequestVariableLifecycle2() throws Exception {
880880

881881
// kick off work (simulating a single request with multiple threads)
882882
for (int t = 0; t < 5; t++) {
883-
Thread th = new Thread(new HystrixContextRunnable(new Runnable() {
883+
Thread th = new Thread(new HystrixContextRunnable(HystrixPlugins.getInstance().getConcurrencyStrategy(), new Runnable() {
884884

885885
@Override
886886
public void run() {

‎hystrix-core/src/main/java/com/netflix/hystrix/HystrixCommand.java

+85-14
Original file line numberDiff line numberDiff line change
@@ -58,6 +58,7 @@
5858
import rx.subjects.ReplaySubject;
5959
import rx.subscriptions.Subscriptions;
6060
import rx.util.functions.Action0;
61+
import rx.util.functions.Action1;
6162
import rx.util.functions.Func1;
6263
import rx.util.functions.Func2;
6364

@@ -1006,6 +1007,24 @@ public Subscription onSubscribe(final Observer<? super R> observer) {
10061007
// TODO better yet, get TimeoutObservable part of Rx
10071008
final SafeObservableSubscription s = new SafeObservableSubscription();
10081009

1010+
/*
1011+
* Define the action to perform on timeout outside of the TimerListener to it can capture the HystrixRequestContext
1012+
* of the calling thread which doesn't exist on the Timer thread.
1013+
*/
1014+
final HystrixContextRunnable timeoutRunnable = new HystrixContextRunnable(originalCommand.concurrencyStrategy, new Runnable() {
1015+
1016+
@Override
1017+
public void run() {
1018+
try {
1019+
R v = originalCommand.getFallbackOrThrowException(HystrixEventType.TIMEOUT, FailureType.TIMEOUT, "timed-out", new TimeoutException());
1020+
observer.onNext(v);
1021+
observer.onCompleted();
1022+
} catch (HystrixRuntimeException re) {
1023+
observer.onError(re);
1024+
}
1025+
}
1026+
});
1027+
10091028
TimerListener listener = new TimerListener() {
10101029

10111030
@Override
@@ -1021,13 +1040,7 @@ public void tick() {
10211040
// we record execution time because we are returning before
10221041
originalCommand.recordTotalExecutionTime(originalCommand.invocationStartTime);
10231042

1024-
try {
1025-
R v = originalCommand.getFallbackOrThrowException(HystrixEventType.TIMEOUT, FailureType.TIMEOUT, "timed-out", new TimeoutException());
1026-
observer.onNext(v);
1027-
observer.onCompleted();
1028-
} catch (HystrixRuntimeException re) {
1029-
observer.onError(re);
1030-
}
1043+
timeoutRunnable.run();
10311044
}
10321045

10331046
s.unsubscribe();
@@ -1146,7 +1159,7 @@ private Subscription subscribeWithThreadIsolation(final Observer<? super R> obse
11461159
}
11471160

11481161
// wrap the synchronous execute() method in a Callable and execute in the threadpool
1149-
final Future<R> f = threadPool.getExecutor().submit(concurrencyStrategy.wrapCallable(new HystrixContextCallable<R>(new Callable<R>() {
1162+
final Future<R> f = threadPool.getExecutor().submit(new HystrixContextCallable<R>(concurrencyStrategy, new Callable<R>() {
11501163

11511164
@Override
11521165
public R call() throws Exception {
@@ -1215,7 +1228,7 @@ private void preTerminationWork(boolean recordDuration) {
12151228
}
12161229
}
12171230

1218-
})));
1231+
}));
12191232

12201233
return new Subscription() {
12211234

@@ -3818,7 +3831,7 @@ public void testExecutionSemaphoreWithQueue() {
38183831
final TryableSemaphore semaphore =
38193832
new TryableSemaphore(HystrixProperty.Factory.asProperty(1));
38203833

3821-
Runnable r = new HystrixContextRunnable(new Runnable() {
3834+
Runnable r = new HystrixContextRunnable(HystrixPlugins.getInstance().getConcurrencyStrategy(), new Runnable() {
38223835

38233836
@Override
38243837
public void run() {
@@ -3890,7 +3903,7 @@ public void testExecutionSemaphoreWithExecution() {
38903903
final TryableSemaphore semaphore =
38913904
new TryableSemaphore(HystrixProperty.Factory.asProperty(1));
38923905

3893-
Runnable r = new HystrixContextRunnable(new Runnable() {
3906+
Runnable r = new HystrixContextRunnable(HystrixPlugins.getInstance().getConcurrencyStrategy(), new Runnable() {
38943907

38953908
@Override
38963909
public void run() {
@@ -3953,7 +3966,7 @@ public void testRejectedExecutionSemaphoreWithFallback() {
39533966

39543967
final AtomicBoolean exceptionReceived = new AtomicBoolean();
39553968

3956-
Runnable r = new HystrixContextRunnable(new Runnable() {
3969+
Runnable r = new HystrixContextRunnable(HystrixPlugins.getInstance().getConcurrencyStrategy(), new Runnable() {
39573970

39583971
@Override
39593972
public void run() {
@@ -4026,7 +4039,7 @@ public void testSemaphorePermitsInUse() {
40264039
// used to signal that all command can finish
40274040
final CountDownLatch sharedLatch = new CountDownLatch(1);
40284041

4029-
final Runnable sharedSemaphoreRunnable = new HystrixContextRunnable(new Runnable() {
4042+
final Runnable sharedSemaphoreRunnable = new HystrixContextRunnable(HystrixPlugins.getInstance().getConcurrencyStrategy(), new Runnable() {
40304043
public void run() {
40314044
try {
40324045
new LatchedSemaphoreCommand(circuitBreaker, sharedSemaphore, startLatch, sharedLatch).execute();
@@ -4054,7 +4067,7 @@ public void run() {
40544067
// tracks failures to obtain semaphores
40554068
final AtomicInteger failureCount = new AtomicInteger();
40564069

4057-
final Thread isolatedThread = new Thread(new HystrixContextRunnable(new Runnable() {
4070+
final Thread isolatedThread = new Thread(new HystrixContextRunnable(HystrixPlugins.getInstance().getConcurrencyStrategy(), new Runnable() {
40584071
public void run() {
40594072
try {
40604073
new LatchedSemaphoreCommand(circuitBreaker, isolatedSemaphore, startLatch, isolatedLatch).execute();
@@ -6133,6 +6146,64 @@ protected String getFallback() {
61336146

61346147
}
61356148

6149+
/**
6150+
* See https://github.com/Netflix/Hystrix/issues/212
6151+
*/
6152+
@Test
6153+
public void testObservableTimeoutNoFallbackThreadContext() {
6154+
final AtomicReference<Thread> onErrorThread = new AtomicReference<Thread>();
6155+
final AtomicBoolean isRequestContextInitialized = new AtomicBoolean();
6156+
TestHystrixCommand<Boolean> command = new TestCommandWithTimeout(50, TestCommandWithTimeout.FALLBACK_NOT_IMPLEMENTED);
6157+
try {
6158+
command.toObservable().doOnError(new Action1<Throwable>() {
6159+
6160+
@Override
6161+
public void call(Throwable t1) {
6162+
System.out.println("onError: " + t1);
6163+
System.out.println("onError Thread: " + Thread.currentThread());
6164+
System.out.println("ThreadContext in onError: " + HystrixRequestContext.isCurrentThreadInitialized());
6165+
onErrorThread.set(Thread.currentThread());
6166+
isRequestContextInitialized.set(HystrixRequestContext.isCurrentThreadInitialized());
6167+
}
6168+
6169+
}).toBlockingObservable().single();
6170+
throw new RuntimeException("expected error to be thrown");
6171+
} catch (Throwable e) {
6172+
assertTrue(isRequestContextInitialized.get());
6173+
assertTrue(onErrorThread.get().getName().startsWith("RxComputationThreadPool"));
6174+
6175+
if (e instanceof HystrixRuntimeException) {
6176+
HystrixRuntimeException de = (HystrixRuntimeException) e;
6177+
assertNotNull(de.getFallbackException());
6178+
assertTrue(de.getFallbackException() instanceof UnsupportedOperationException);
6179+
assertNotNull(de.getImplementingClass());
6180+
assertNotNull(de.getCause());
6181+
assertTrue(de.getCause() instanceof TimeoutException);
6182+
} else {
6183+
fail("the exception should be ExecutionException with cause as HystrixRuntimeException");
6184+
}
6185+
}
6186+
6187+
assertTrue(command.getExecutionTimeInMilliseconds() > -1);
6188+
assertTrue(command.isResponseTimedOut());
6189+
6190+
assertEquals(0, command.builder.metrics.getRollingCount(HystrixRollingNumberEvent.SUCCESS));
6191+
assertEquals(1, command.builder.metrics.getRollingCount(HystrixRollingNumberEvent.EXCEPTION_THROWN));
6192+
assertEquals(0, command.builder.metrics.getRollingCount(HystrixRollingNumberEvent.FAILURE));
6193+
assertEquals(0, command.builder.metrics.getRollingCount(HystrixRollingNumberEvent.FALLBACK_REJECTION));
6194+
assertEquals(0, command.builder.metrics.getRollingCount(HystrixRollingNumberEvent.FALLBACK_FAILURE));
6195+
assertEquals(0, command.builder.metrics.getRollingCount(HystrixRollingNumberEvent.FALLBACK_SUCCESS));
6196+
assertEquals(0, command.builder.metrics.getRollingCount(HystrixRollingNumberEvent.SEMAPHORE_REJECTED));
6197+
assertEquals(0, command.builder.metrics.getRollingCount(HystrixRollingNumberEvent.SHORT_CIRCUITED));
6198+
assertEquals(0, command.builder.metrics.getRollingCount(HystrixRollingNumberEvent.THREAD_POOL_REJECTED));
6199+
assertEquals(1, command.builder.metrics.getRollingCount(HystrixRollingNumberEvent.TIMEOUT));
6200+
assertEquals(0, command.builder.metrics.getRollingCount(HystrixRollingNumberEvent.RESPONSE_FROM_CACHE));
6201+
6202+
assertEquals(100, command.builder.metrics.getHealthCounts().getErrorPercentage());
6203+
6204+
assertEquals(1, HystrixRequestLog.getCurrentRequest().getExecutedCommands().size());
6205+
}
6206+
61366207
/* ******************************************************************************** */
61376208
/* ******************************************************************************** */
61386209
/* private HystrixCommand class implementations for unit testing */

‎hystrix-core/src/main/java/com/netflix/hystrix/collapser/RequestCollapser.java

+2-2
Original file line numberDiff line numberDiff line change
@@ -121,7 +121,7 @@ private class CollapsedTask implements TimerListener {
121121
CollapsedTask() {
122122
// this gets executed from the context of a HystrixCommand parent thread (such as a Tomcat thread)
123123
// so we create the callable now where we can capture the thread context
124-
callableWithContextOfParent = concurrencyStrategy.wrapCallable(new HystrixContextCallable<Void>(new Callable<Void>() {
124+
callableWithContextOfParent = new HystrixContextCallable<Void>(concurrencyStrategy, new Callable<Void>() {
125125
// the wrapCallable call allows a strategy to capture thread-context if desired
126126

127127
@Override
@@ -144,7 +144,7 @@ public Void call() throws Exception {
144144
return null;
145145
}
146146

147-
}));
147+
});
148148
}
149149

150150
@Override

‎hystrix-core/src/main/java/com/netflix/hystrix/strategy/HystrixPlugins.java

+77
Original file line numberDiff line numberDiff line change
@@ -17,13 +17,20 @@
1717

1818
import static org.junit.Assert.*;
1919

20+
import java.util.concurrent.Callable;
2021
import java.util.concurrent.atomic.AtomicReference;
2122

2223
import org.junit.After;
2324
import org.junit.Test;
2425

26+
import rx.util.functions.Action1;
27+
28+
import com.netflix.hystrix.Hystrix;
29+
import com.netflix.hystrix.HystrixCommand;
30+
import com.netflix.hystrix.HystrixCommandGroupKey;
2531
import com.netflix.hystrix.strategy.concurrency.HystrixConcurrencyStrategy;
2632
import com.netflix.hystrix.strategy.concurrency.HystrixConcurrencyStrategyDefault;
33+
import com.netflix.hystrix.strategy.concurrency.HystrixRequestContext;
2734
import com.netflix.hystrix.strategy.eventnotifier.HystrixEventNotifier;
2835
import com.netflix.hystrix.strategy.eventnotifier.HystrixEventNotifierDefault;
2936
import com.netflix.hystrix.strategy.executionhook.HystrixCommandExecutionHook;
@@ -418,6 +425,76 @@ public static class HystrixPropertiesStrategyTestImpl extends HystrixPropertiesS
418425
private static String getFullClassNameForTestClass(Class<?> cls) {
419426
return HystrixPlugins.class.getPackage().getName() + "." + HystrixPlugins.class.getSimpleName() + "$UnitTest$" + cls.getSimpleName();
420427
}
428+
429+
430+
private static final ThreadLocal<String> testRequestIdThreadLocal = new ThreadLocal<String>();
431+
432+
public static class DummyCommand extends HystrixCommand<Void> {
433+
434+
public DummyCommand() {
435+
super(HystrixCommandGroupKey.Factory.asKey("Dummy"));
436+
}
437+
438+
@Override
439+
protected Void run() throws Exception {
440+
System.out.println("requestId (run) = " + testRequestIdThreadLocal.get());
441+
Thread.sleep(2000);
442+
return null;
443+
}
444+
}
445+
446+
@Test
447+
public void testRequestContextViaPluginInTimeout() {
448+
HystrixPlugins.getInstance().registerConcurrencyStrategy(new HystrixConcurrencyStrategy() {
449+
@Override
450+
public <T> Callable<T> wrapCallable(final Callable<T> callable) {
451+
return new RequestIdCallable<T>(callable);
452+
}
453+
});
454+
455+
HystrixRequestContext context = HystrixRequestContext.initializeContext();
456+
457+
testRequestIdThreadLocal.set("foobar");
458+
final AtomicReference<String> valueInTimeout = new AtomicReference<String>();
459+
460+
new DummyCommand().toObservable()
461+
.doOnError(new Action1<Throwable>() {
462+
@Override
463+
public void call(Throwable throwable) {
464+
System.out.println("initialized = " + HystrixRequestContext.isCurrentThreadInitialized());
465+
System.out.println("requestId (timeout) = " + testRequestIdThreadLocal.get());
466+
valueInTimeout.set(testRequestIdThreadLocal.get());
467+
}
468+
})
469+
.materialize()
470+
.toBlockingObservable().single();
471+
472+
context.shutdown();
473+
Hystrix.reset();
474+
475+
assertEquals("foobar", valueInTimeout.get());
476+
}
477+
478+
private static class RequestIdCallable<T> implements Callable<T> {
479+
private final Callable<T> callable;
480+
private final String requestId;
481+
482+
public RequestIdCallable(Callable<T> callable) {
483+
this.callable = callable;
484+
this.requestId = testRequestIdThreadLocal.get();
485+
}
486+
487+
@Override
488+
public T call() throws Exception {
489+
String original = testRequestIdThreadLocal.get();
490+
testRequestIdThreadLocal.set(requestId);
491+
try {
492+
return callable.call();
493+
} finally {
494+
testRequestIdThreadLocal.set(original);
495+
}
496+
}
497+
}
421498
}
422499

423500
}

‎hystrix-core/src/main/java/com/netflix/hystrix/strategy/concurrency/HystrixConcurrencyStrategy.java

+2
Original file line numberDiff line numberDiff line change
@@ -32,6 +32,7 @@
3232
import rx.util.functions.Func1;
3333

3434
import com.netflix.config.ConfigurationManager;
35+
import com.netflix.hystrix.Hystrix;
3536
import com.netflix.hystrix.HystrixCommand;
3637
import com.netflix.hystrix.HystrixCommandGroupKey;
3738
import com.netflix.hystrix.HystrixRequestLog;
@@ -218,6 +219,7 @@ protected String run() throws Exception {
218219
}
219220

220221
}
222+
221223
}
222224

223225
}

‎hystrix-core/src/main/java/com/netflix/hystrix/strategy/concurrency/HystrixContextCallable.java

+2-2
Original file line numberDiff line numberDiff line change
@@ -30,8 +30,8 @@ public class HystrixContextCallable<K> implements Callable<K> {
3030
private final Callable<K> actual;
3131
private final HystrixRequestContext parentThreadState;
3232

33-
public HystrixContextCallable(Callable<K> actual) {
34-
this.actual = actual;
33+
public HystrixContextCallable(HystrixConcurrencyStrategy concurrencyStrategy, Callable<K> actual) {
34+
this.actual = concurrencyStrategy.wrapCallable(actual);
3535
this.parentThreadState = HystrixRequestContext.getContextForCurrentThread();
3636
}
3737

‎hystrix-core/src/main/java/com/netflix/hystrix/strategy/concurrency/HystrixContextRunnable.java

+18-4
Original file line numberDiff line numberDiff line change
@@ -15,18 +15,28 @@
1515
*/
1616
package com.netflix.hystrix.strategy.concurrency;
1717

18+
import java.util.concurrent.Callable;
19+
1820
/**
1921
* Wrapper around {@link Runnable} that manages the {@link HystrixRequestContext} initialization and cleanup for the execution of the {@link Runnable}
2022
*
2123
* @ExcludeFromJavadoc
2224
*/
2325
public class HystrixContextRunnable implements Runnable {
2426

25-
private final Runnable actual;
27+
private final Callable<Void> actual;
2628
private final HystrixRequestContext parentThreadState;
2729

28-
public HystrixContextRunnable(Runnable actual) {
29-
this.actual = actual;
30+
public HystrixContextRunnable(HystrixConcurrencyStrategy concurrencyStrategy, final Runnable actual) {
31+
this.actual = concurrencyStrategy.wrapCallable(new Callable<Void>() {
32+
33+
@Override
34+
public Void call() throws Exception {
35+
actual.run();
36+
return null;
37+
}
38+
39+
});
3040
this.parentThreadState = HystrixRequestContext.getContextForCurrentThread();
3141
}
3242

@@ -37,7 +47,11 @@ public void run() {
3747
// set the state of this thread to that of its parent
3848
HystrixRequestContext.setContextOnCurrentThread(parentThreadState);
3949
// execute actual Callable with the state of the parent
40-
actual.run();
50+
try {
51+
actual.call();
52+
} catch (Exception e) {
53+
throw new RuntimeException(e);
54+
}
4155
} finally {
4256
// restore this thread back to its original state
4357
HystrixRequestContext.setContextOnCurrentThread(existingState);

‎hystrix-core/src/main/java/com/netflix/hystrix/strategy/concurrency/HystrixContextScheduler.java

-2
Original file line numberDiff line numberDiff line change
@@ -15,12 +15,10 @@
1515
*/
1616
package com.netflix.hystrix.strategy.concurrency;
1717

18-
import java.util.concurrent.Callable;
1918
import java.util.concurrent.TimeUnit;
2019

2120
import rx.Scheduler;
2221
import rx.Subscription;
23-
import rx.subscriptions.MultipleAssignmentSubscription;
2422
import rx.util.functions.Func2;
2523

2624
/**

0 commit comments

Comments
 (0)