diff --git a/hystrix-core/src/main/java/com/netflix/hystrix/AbstractCommand.java b/hystrix-core/src/main/java/com/netflix/hystrix/AbstractCommand.java index db3658d76..e0d5ad06d 100644 --- a/hystrix-core/src/main/java/com/netflix/hystrix/AbstractCommand.java +++ b/hystrix-core/src/main/java/com/netflix/hystrix/AbstractCommand.java @@ -383,11 +383,8 @@ public void call(Subscriber observer) { metrics.incrementConcurrentExecutionCount(); // mark that we're starting execution on the ExecutionHook - try { - executionHook.onStart(_this); - } catch (Throwable hookEx) { - logger.warn("Error calling HystrixCommandExecutionHook.onStart", hookEx); - } + // if this hook throws an exception, then a fast-fail occurs with no fallback. No state is left inconsistent + executionHook.onStart(_this); /* determine if we're allowed to execute */ if (circuitBreaker.allowRequest()) { @@ -519,26 +516,21 @@ public void call(Subscriber s) { s.onError(new RuntimeException("timed out before executing run()")); } else { // not timed out so execute - try { - executionHook.onThreadStart(_self); - } catch (Throwable hookEx) { - logger.warn("Error calling HystrixCommandExecutionHook.onThreadStart", hookEx); - } - try { - executionHook.onRunStart(_self); - } catch (Throwable hookEx) { - logger.warn("Error calling HystrixCommandExecutionHook.onRunStart", hookEx); - } - try { - executionHook.onExecutionStart(_self); - } catch (Throwable hookEx) { - logger.warn("Error calling HystrixCommandExecutionHook.onExecutionStart", hookEx); - } HystrixCounters.incrementGlobalConcurrentThreads(); threadPool.markThreadExecution(); // store the command that is being run endCurrentThreadExecutingCommand.set(Hystrix.startCurrentThreadExecutingCommand(getCommandKey())); isExecutedInThread.set(true); + /** + * If any of these hooks throw an exception, then it appears as if the actual execution threw an error + */ + try { + executionHook.onThreadStart(_self); + executionHook.onRunStart(_self); + executionHook.onExecutionStart(_self); + } catch (Throwable ex) { + s.onError(ex); + } getExecutionObservableWithLifecycle().unsafeSubscribe(s); //the getExecutionObservableWithLifecycle method already wraps sync exceptions, so no need to catch here } } @@ -551,19 +543,16 @@ public Boolean call() { })); } else { // semaphore isolated + // store the command that is being run + endCurrentThreadExecutingCommand.set(Hystrix.startCurrentThreadExecutingCommand(getCommandKey())); try { executionHook.onRunStart(_self); - } catch (Throwable hookEx) { - logger.warn("Error calling HystrixCommandExecutionHook.onRunStart", hookEx); - } - try { executionHook.onExecutionStart(_self); - } catch (Throwable hookEx) { - logger.warn("Error calling HystrixCommandExecutionHook.onExecutionStart", hookEx); + run = getExecutionObservableWithLifecycle(); //the getExecutionObservableWithLifecycle method already wraps sync exceptions, so this shouldn't throw + } catch (Throwable ex) { + //If the above hooks throw, then use that as the result of the run method + run = Observable.error(ex); } - // store the command that is being run - endCurrentThreadExecutingCommand.set(Hystrix.startCurrentThreadExecutingCommand(getCommandKey())); - run = getExecutionObservableWithLifecycle(); //the getExecutionObservableWithLifecycle method already wraps sync exceptions, so no need to catch here } run = run.doOnEach(new Action1>() { @@ -753,19 +742,17 @@ private Observable getFallbackOrThrowException(final HystrixEventType eventTy // acquire a permit if (fallbackSemaphore.tryAcquire()) { - if (isFallbackUserSupplied(this)) { - try { + try { + if (isFallbackUserSupplied(this)) { executionHook.onFallbackStart(this); - } catch (Throwable hookEx) { - logger.warn("Error calling HystrixCommandExecutionHook.onFallbackStart", hookEx); + fallbackExecutionChain = getFallbackObservable(); + } else { + //same logic as above without the hook invocation + fallbackExecutionChain = getFallbackObservable(); } - } - - try { - fallbackExecutionChain = getFallbackObservable(); - } catch (Throwable t) { - // getFallback() is user provided and can throw so we catch it and turn it into Observable.error - fallbackExecutionChain = Observable.error(t); + } catch(Throwable ex) { + //If hook or user-fallback throws, then use that as the result of the fallback lookup + fallbackExecutionChain = Observable.error(ex); } fallbackExecutionChain = fallbackExecutionChain diff --git a/hystrix-core/src/main/java/com/netflix/hystrix/util/HystrixRollingPercentile.java b/hystrix-core/src/main/java/com/netflix/hystrix/util/HystrixRollingPercentile.java index 9e581a533..afb267bf6 100644 --- a/hystrix-core/src/main/java/com/netflix/hystrix/util/HystrixRollingPercentile.java +++ b/hystrix-core/src/main/java/com/netflix/hystrix/util/HystrixRollingPercentile.java @@ -19,8 +19,6 @@ import java.util.Arrays; import java.util.Collections; import java.util.Iterator; -import java.util.concurrent.atomic.AtomicInteger; -import java.util.concurrent.atomic.AtomicIntegerArray; import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.atomic.AtomicReferenceArray; import java.util.concurrent.locks.ReentrantLock; diff --git a/hystrix-core/src/test/java/com/netflix/hystrix/HystrixCommandTest.java b/hystrix-core/src/test/java/com/netflix/hystrix/HystrixCommandTest.java index ffa08b087..ac1c7cedc 100644 --- a/hystrix-core/src/test/java/com/netflix/hystrix/HystrixCommandTest.java +++ b/hystrix-core/src/test/java/com/netflix/hystrix/HystrixCommandTest.java @@ -36,6 +36,7 @@ import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; +import com.netflix.hystrix.strategy.executionhook.HystrixCommandExecutionHook; import org.junit.After; import org.junit.Before; import org.junit.Test; @@ -4206,6 +4207,68 @@ protected Integer getFallback() { assertTrue(1 == new PrimaryCommand(new TestCircuitBreaker()).execute()); } + @Test + public void testOnRunStartHookThrows() { + final AtomicBoolean threadExceptionEncountered = new AtomicBoolean(false); + final AtomicBoolean semaphoreExceptionEncountered = new AtomicBoolean(false); + final AtomicBoolean onThreadStartInvoked = new AtomicBoolean(false); + final AtomicBoolean onThreadCompleteInvoked = new AtomicBoolean(false); + + class FailureInjectionHook extends HystrixCommandExecutionHook { + @Override + public void onExecutionStart(HystrixInvokable commandInstance) { + throw new HystrixRuntimeException(HystrixRuntimeException.FailureType.COMMAND_EXCEPTION, commandInstance.getClass(), "Injected Failure", null, null); + } + + @Override + public void onThreadStart(HystrixInvokable commandInstance) { + onThreadStartInvoked.set(true); + super.onThreadStart(commandInstance); + } + + @Override + public void onThreadComplete(HystrixInvokable commandInstance) { + onThreadCompleteInvoked.set(true); + super.onThreadComplete(commandInstance); + } + } + + final FailureInjectionHook failureInjectionHook = new FailureInjectionHook(); + + class FailureInjectedCommand extends TestHystrixCommand { + public FailureInjectedCommand(ExecutionIsolationStrategy isolationStrategy) { + super(testPropsBuilder().setCommandPropertiesDefaults(HystrixCommandPropertiesTest.getUnitTestPropertiesSetter().withExecutionIsolationStrategy(isolationStrategy)), failureInjectionHook); + } + + @Override + protected Integer run() throws Exception { + return 3; + } + } + + TestHystrixCommand threadCmd = new FailureInjectedCommand(ExecutionIsolationStrategy.THREAD); + try { + int result = threadCmd.execute(); + System.out.println("RESULT : " + result); + } catch (Throwable ex) { + ex.printStackTrace(); + threadExceptionEncountered.set(true); + } + assertTrue(threadExceptionEncountered.get()); + assertTrue(onThreadStartInvoked.get()); + assertTrue(onThreadCompleteInvoked.get()); + + TestHystrixCommand semaphoreCmd = new FailureInjectedCommand(ExecutionIsolationStrategy.SEMAPHORE); + try { + int result = semaphoreCmd.execute(); + System.out.println("RESULT : " + result); + } catch (Throwable ex) { + ex.printStackTrace(); + semaphoreExceptionEncountered.set(true); + } + assertTrue(semaphoreExceptionEncountered.get()); + } + /* ******************************************************************************** */ /* ******************************************************************************** */ /* private HystrixCommand class implementations for unit testing */ diff --git a/hystrix-core/src/test/java/com/netflix/hystrix/TestHystrixCommand.java b/hystrix-core/src/test/java/com/netflix/hystrix/TestHystrixCommand.java index 03920c2bf..872b4e863 100644 --- a/hystrix-core/src/test/java/com/netflix/hystrix/TestHystrixCommand.java +++ b/hystrix-core/src/test/java/com/netflix/hystrix/TestHystrixCommand.java @@ -15,6 +15,8 @@ */ package com.netflix.hystrix; +import com.netflix.hystrix.strategy.executionhook.HystrixCommandExecutionHook; + abstract public class TestHystrixCommand extends HystrixCommand implements AbstractTestHystrixCommand { private final TestCommandBuilder builder; @@ -26,6 +28,13 @@ public TestHystrixCommand(TestCommandBuilder builder) { this.builder = builder; } + public TestHystrixCommand(TestCommandBuilder builder, HystrixCommandExecutionHook executionHook) { + super(builder.owner, builder.dependencyKey, builder.threadPoolKey, builder.circuitBreaker, builder.threadPool, + builder.commandPropertiesDefaults, builder.threadPoolPropertiesDefaults, builder.metrics, + builder.fallbackSemaphore, builder.executionSemaphore, TEST_PROPERTIES_FACTORY, executionHook); + this.builder = builder; + } + public TestCommandBuilder getBuilder() { return builder; }