HystrixCommand extends AbstractCommand<R> implements HystrixExecutable<R>, HystrixInvokableInfo<R>, HystrixObservable<R> { public Future<R> queue() { final Future<R> delegate = toObservable().toBlocking().toFuture(); final Future<R> f = new Future<R>() { @Override public boolean cancel(boolean mayInterruptIfRunning) { if (delegate.isCancelled()) { return false; } if (HystrixCommand.this.getProperties().executionIsolationThreadInterruptOnFutureCancel().get()) { interruptOnFutureCancel.compareAndSet(false, mayInterruptIfRunning); } final boolean res = delegate.cancel(interruptOnFutureCancel.get()); if (!isExecutionComplete() && interruptOnFutureCancel.get()) { final Thread t = executionThread.get(); if (t != null && !t.equals(Thread.currentThread())) { t.interrupt(); } } return res; } @Override public boolean isCancelled() { return delegate.isCancelled(); } @Override public boolean isDone() { return delegate.isDone(); } @Override public R get() throws InterruptedException, ExecutionException { return delegate.get(); } @Override public R get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException { return delegate.get(timeout, unit); } }; if (f.isDone()) { try { f.get(); return f; } catch (Exception e) { Throwable t = decomposeException(e); if (t instanceof HystrixBadRequestException) { return f; } else if (t instanceof HystrixRuntimeException) { HystrixRuntimeException hre = (HystrixRuntimeException) t; switch (hre.getFailureType()) { case COMMAND_EXCEPTION: case TIMEOUT: return f; default: throw hre; } } else { throw Exceptions.sneakyThrow(t); } } } return f; } protected  HystrixCommand(HystrixCommandGroupKey group); protected  HystrixCommand(HystrixCommandGroupKey group, HystrixThreadPoolKey threadPool); protected  HystrixCommand(HystrixCommandGroupKey group, int executionIsolationThreadTimeoutInMilliseconds); protected  HystrixCommand(HystrixCommandGroupKey group, HystrixThreadPoolKey threadPool, int executionIsolationThreadTimeoutInMilliseconds); protected  HystrixCommand(Setter setter);   HystrixCommand(HystrixCommandGroupKey group, HystrixCommandKey key, HystrixThreadPoolKey threadPoolKey, HystrixCircuitBreaker circuitBreaker, HystrixThreadPool threadPool,
            HystrixCommandProperties.Setter commandPropertiesDefaults, HystrixThreadPoolProperties.Setter threadPoolPropertiesDefaults,
            HystrixCommandMetrics metrics, TryableSemaphore fallbackSemaphore, TryableSemaphore executionSemaphore,
            HystrixPropertiesStrategy propertiesStrategy, HystrixCommandExecutionHook executionHook); R execute(); Future<R> queue();  }
@Test public void testQueueSuccess() throws Exception { TestHystrixCommand<Integer> command = getCommand(ExecutionIsolationStrategy.THREAD, AbstractTestHystrixCommand.ExecutionResult.SUCCESS); Future<Integer> future = command.queue(); assertEquals(FlexibleTestHystrixCommand.EXECUTE_VALUE, future.get()); assertTrue(command.getExecutionTimeInMilliseconds() > -1); assertTrue(command.isSuccessfulExecution()); assertCommandExecutionEvents(command, HystrixEventType.SUCCESS); assertNull(command.getExecutionException()); assertEquals(0, command.getBuilder().metrics.getCurrentConcurrentExecutionCount()); assertSaneHystrixRequestLog(1); }
@Test public void testQueueKnownFailureWithNoFallback() { TestHystrixCommand<Integer> command = getCommand(ExecutionIsolationStrategy.THREAD, AbstractTestHystrixCommand.ExecutionResult.HYSTRIX_FAILURE, AbstractTestHystrixCommand.FallbackResult.UNIMPLEMENTED); try { command.queue().get(); fail("we shouldn't get here"); } catch (Exception e) { e.printStackTrace(); if (e.getCause() instanceof HystrixRuntimeException) { HystrixRuntimeException de = (HystrixRuntimeException) e.getCause(); assertNotNull(de.getFallbackException()); assertNotNull(de.getImplementingClass()); } else { fail("the cause should be HystrixRuntimeException"); } } assertTrue(command.getExecutionTimeInMilliseconds() > -1); assertTrue(command.isFailedExecution()); assertCommandExecutionEvents(command, HystrixEventType.FAILURE, HystrixEventType.FALLBACK_MISSING); assertNotNull(command.getExecutionException()); assertEquals(0, command.getBuilder().metrics.getCurrentConcurrentExecutionCount()); assertSaneHystrixRequestLog(1); }
@Test public void testQueueUnknownFailureWithNoFallback() { TestHystrixCommand<Integer> command = getCommand(ExecutionIsolationStrategy.THREAD, AbstractTestHystrixCommand.ExecutionResult.FAILURE, AbstractTestHystrixCommand.FallbackResult.UNIMPLEMENTED); try { command.queue().get(); fail("we shouldn't get here"); } catch (Exception e) { e.printStackTrace(); if (e.getCause() instanceof HystrixRuntimeException) { HystrixRuntimeException de = (HystrixRuntimeException) e.getCause(); assertNotNull(de.getFallbackException()); assertNotNull(de.getImplementingClass()); } else { fail("the cause should be HystrixRuntimeException"); } } assertTrue(command.getExecutionTimeInMilliseconds() > -1); assertTrue(command.isFailedExecution()); assertCommandExecutionEvents(command, HystrixEventType.FAILURE, HystrixEventType.FALLBACK_MISSING); assertNotNull(command.getExecutionException()); assertEquals(0, command.getBuilder().metrics.getCurrentConcurrentExecutionCount()); assertSaneHystrixRequestLog(1); }
@Test public void testQueueFailureWithFallback() { TestHystrixCommand<Integer> command = getCommand(ExecutionIsolationStrategy.THREAD, AbstractTestHystrixCommand.ExecutionResult.FAILURE, AbstractTestHystrixCommand.FallbackResult.SUCCESS); try { Future<Integer> future = command.queue(); assertEquals(FlexibleTestHystrixCommand.FALLBACK_VALUE, future.get()); } catch (Exception e) { e.printStackTrace(); fail("We should have received a response from the fallback."); } assertTrue(command.getExecutionTimeInMilliseconds() > -1); assertTrue(command.isFailedExecution()); assertCommandExecutionEvents(command, HystrixEventType.FAILURE, HystrixEventType.FALLBACK_SUCCESS); assertNotNull(command.getExecutionException()); assertEquals(0, command.getBuilder().metrics.getCurrentConcurrentExecutionCount()); assertSaneHystrixRequestLog(1); }
@Test public void testQueueFailureWithFallbackFailure() { TestHystrixCommand<Integer> command = getCommand(ExecutionIsolationStrategy.THREAD, AbstractTestHystrixCommand.ExecutionResult.FAILURE, AbstractTestHystrixCommand.FallbackResult.FAILURE); try { command.queue().get(); fail("we shouldn't get here"); } catch (Exception e) { if (e.getCause() instanceof HystrixRuntimeException) { HystrixRuntimeException de = (HystrixRuntimeException) e.getCause(); e.printStackTrace(); assertNotNull(de.getFallbackException()); } else { fail("the cause should be HystrixRuntimeException"); } } assertTrue(command.getExecutionTimeInMilliseconds() > -1); assertTrue(command.isFailedExecution()); assertCommandExecutionEvents(command, HystrixEventType.FAILURE, HystrixEventType.FALLBACK_FAILURE); assertNotNull(command.getExecutionException()); assertEquals(0, command.getBuilder().metrics.getCurrentConcurrentExecutionCount()); assertSaneHystrixRequestLog(1); }
@Test public void testQueuedExecutionTimeoutWithNoFallback() { TestHystrixCommand<Integer> command = getCommand(ExecutionIsolationStrategy.THREAD, AbstractTestHystrixCommand.ExecutionResult.SUCCESS, 200, AbstractTestHystrixCommand.FallbackResult.UNIMPLEMENTED, 50); try { command.queue().get(); fail("we shouldn't get here"); } catch (Exception e) { e.printStackTrace(); if (e instanceof ExecutionException && e.getCause() instanceof HystrixRuntimeException) { HystrixRuntimeException de = (HystrixRuntimeException) e.getCause(); assertNotNull(de.getFallbackException()); assertTrue(de.getFallbackException() instanceof UnsupportedOperationException); assertNotNull(de.getImplementingClass()); assertNotNull(de.getCause()); assertTrue(de.getCause() instanceof TimeoutException); } else { fail("the exception should be ExecutionException with cause as HystrixRuntimeException"); } } assertTrue(command.getExecutionTimeInMilliseconds() > -1); assertTrue(command.isResponseTimedOut()); assertCommandExecutionEvents(command, HystrixEventType.TIMEOUT, HystrixEventType.FALLBACK_MISSING); assertNotNull(command.getExecutionException()); assertEquals(0, command.getBuilder().metrics.getCurrentConcurrentExecutionCount()); assertSaneHystrixRequestLog(1); }
@Test public void testQueuedExecutionTimeoutWithFallback() throws Exception { TestHystrixCommand<Integer> command = getCommand(ExecutionIsolationStrategy.THREAD, AbstractTestHystrixCommand.ExecutionResult.SUCCESS, 200, AbstractTestHystrixCommand.FallbackResult.SUCCESS, 50); assertEquals(FlexibleTestHystrixCommand.FALLBACK_VALUE, command.queue().get()); assertCommandExecutionEvents(command, HystrixEventType.TIMEOUT, HystrixEventType.FALLBACK_SUCCESS); assertNotNull(command.getExecutionException()); assertEquals(0, command.getBuilder().metrics.getCurrentConcurrentExecutionCount()); assertSaneHystrixRequestLog(1); }
@Test public void testQueuedExecutionTimeoutFallbackFailure() { TestHystrixCommand<Integer> command = getCommand(ExecutionIsolationStrategy.THREAD, AbstractTestHystrixCommand.ExecutionResult.SUCCESS, 200, AbstractTestHystrixCommand.FallbackResult.FAILURE, 50); try { command.queue().get(); fail("we shouldn't get here"); } catch (Exception e) { if (e instanceof ExecutionException && e.getCause() instanceof HystrixRuntimeException) { HystrixRuntimeException de = (HystrixRuntimeException) e.getCause(); assertNotNull(de.getFallbackException()); assertFalse(de.getFallbackException() instanceof UnsupportedOperationException); assertNotNull(de.getImplementingClass()); assertNotNull(de.getCause()); assertTrue(de.getCause() instanceof TimeoutException); } else { fail("the exception should be ExecutionException with cause as HystrixRuntimeException"); } } assertCommandExecutionEvents(command, HystrixEventType.TIMEOUT, HystrixEventType.FALLBACK_FAILURE); assertNotNull(command.getExecutionException()); assertEquals(0, command.getBuilder().metrics.getCurrentConcurrentExecutionCount()); assertSaneHystrixRequestLog(1); }
@Test public void testRejectedThreadWithNoFallback() throws Exception { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("Rejection-NoFallback"); TestCircuitBreaker circuitBreaker = new TestCircuitBreaker(); SingleThreadedPoolWithQueue pool = new SingleThreadedPoolWithQueue(1); pool.queue.add(new Runnable() { @Override public void run() { System.out.println("**** queue filler1 ****"); try { Thread.sleep(500); } catch (InterruptedException e) { e.printStackTrace(); } } }); Future<Boolean> f = null; TestCommandRejection command1 = null; TestCommandRejection command2 = null; try { command1 = new TestCommandRejection(key, circuitBreaker, pool, 500, 600, TestCommandRejection.FALLBACK_NOT_IMPLEMENTED); f = command1.queue(); command2 = new TestCommandRejection(key, circuitBreaker, pool, 500, 600, TestCommandRejection.FALLBACK_NOT_IMPLEMENTED); command2.queue(); fail("we shouldn't get here"); } catch (Exception e) { e.printStackTrace(); System.out.println("command.getExecutionTimeInMilliseconds(): " + command2.getExecutionTimeInMilliseconds()); assertTrue(command2.isResponseRejected()); assertFalse(command2.isResponseShortCircuited()); assertFalse(command2.isResponseTimedOut()); assertNotNull(command2.getExecutionException()); if (e instanceof HystrixRuntimeException && e.getCause() instanceof RejectedExecutionException) { HystrixRuntimeException de = (HystrixRuntimeException) e; assertNotNull(de.getFallbackException()); assertTrue(de.getFallbackException() instanceof UnsupportedOperationException); assertNotNull(de.getImplementingClass()); assertNotNull(de.getCause()); assertTrue(de.getCause() instanceof RejectedExecutionException); } else { fail("the exception should be HystrixRuntimeException with cause as RejectedExecutionException"); } } f.get(); assertCommandExecutionEvents(command1, HystrixEventType.SUCCESS); assertCommandExecutionEvents(command2, HystrixEventType.THREAD_POOL_REJECTED, HystrixEventType.FALLBACK_MISSING); assertEquals(0, circuitBreaker.metrics.getCurrentConcurrentExecutionCount()); assertSaneHystrixRequestLog(2); }
@Test public void testRejectedThreadWithFallbackFailure() throws ExecutionException, InterruptedException { TestCircuitBreaker circuitBreaker = new TestCircuitBreaker(); SingleThreadedPoolWithQueue pool = new SingleThreadedPoolWithQueue(1); HystrixCommandKey key = HystrixCommandKey.Factory.asKey("Rejection-A"); TestCommandRejection command1 = new TestCommandRejection(key, circuitBreaker, pool, 500, 600, TestCommandRejection.FALLBACK_FAILURE); TestCommandRejection command2 = new TestCommandRejection(key, circuitBreaker, pool, 500, 600, TestCommandRejection.FALLBACK_SUCCESS); TestCommandRejection command3 = new TestCommandRejection(key, circuitBreaker, pool, 500, 600, TestCommandRejection.FALLBACK_FAILURE); Future<Boolean> f1 = null; Future<Boolean> f2 = null; try { f1 = command1.queue(); f2 = command2.queue(); assertEquals(false, command3.queue().get()); fail("we shouldn't get here"); } catch (Exception e) { e.printStackTrace(); if (e instanceof HystrixRuntimeException && e.getCause() instanceof RejectedExecutionException) { HystrixRuntimeException de = (HystrixRuntimeException) e; assertNotNull(de.getFallbackException()); assertFalse(de.getFallbackException() instanceof UnsupportedOperationException); assertNotNull(de.getImplementingClass()); assertNotNull(de.getCause()); assertTrue(de.getCause() instanceof RejectedExecutionException); } else { fail("the exception should be HystrixRuntimeException with cause as RejectedExecutionException"); } } assertCommandExecutionEvents(command1); assertCommandExecutionEvents(command2); assertCommandExecutionEvents(command3, HystrixEventType.THREAD_POOL_REJECTED, HystrixEventType.FALLBACK_FAILURE); int numInFlight = circuitBreaker.metrics.getCurrentConcurrentExecutionCount(); assertTrue("Expected at most 1 in flight but got : " + numInFlight, numInFlight <= 1); HystrixRequestLog requestLog = HystrixRequestLog.getCurrentRequest(); assertEquals(3, requestLog.getAllExecutedCommands().size()); assertTrue(requestLog.getExecutedCommandsAsString().contains("Executed")); long startTime = System.currentTimeMillis(); f1.get(); f2.get(); assertEquals(0, circuitBreaker.metrics.getCurrentConcurrentExecutionCount()); System.out.println("Time blocked : " + (System.currentTimeMillis() - startTime)); }
@Test public void testRejectedThreadUsingQueueSize() { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("Rejection-B"); TestCircuitBreaker circuitBreaker = new TestCircuitBreaker(); SingleThreadedPoolWithQueue pool = new SingleThreadedPoolWithQueue(10, 1); pool.queue.add(new Runnable() { @Override public void run() { System.out.println("**** queue filler1 ****"); try { Thread.sleep(500); } catch (InterruptedException e) { e.printStackTrace(); } } }); TestCommandRejection command = new TestCommandRejection(key, circuitBreaker, pool, 500, 600, TestCommandRejection.FALLBACK_NOT_IMPLEMENTED); try { command.queue(); fail("we shouldn't get here"); } catch (Exception e) { e.printStackTrace(); assertTrue(command.isResponseRejected()); assertFalse(command.isResponseShortCircuited()); assertFalse(command.isResponseTimedOut()); assertNotNull(command.getExecutionException()); if (e instanceof HystrixRuntimeException && e.getCause() instanceof RejectedExecutionException) { HystrixRuntimeException de = (HystrixRuntimeException) e; assertNotNull(de.getFallbackException()); assertTrue(de.getFallbackException() instanceof UnsupportedOperationException); assertNotNull(de.getImplementingClass()); assertNotNull(de.getCause()); assertTrue(de.getCause() instanceof RejectedExecutionException); } else { fail("the exception should be HystrixRuntimeException with cause as RejectedExecutionException"); } } assertCommandExecutionEvents(command, HystrixEventType.THREAD_POOL_REJECTED, HystrixEventType.FALLBACK_MISSING); assertEquals(0, circuitBreaker.metrics.getCurrentConcurrentExecutionCount()); assertSaneHystrixRequestLog(1); }
@Test public void testFallbackSemaphore() throws Exception { TestCircuitBreaker circuitBreaker = new TestCircuitBreaker(); TestSemaphoreCommandWithSlowFallback command1 = new TestSemaphoreCommandWithSlowFallback(circuitBreaker, 1, 200); boolean result = command1.queue().get(); assertTrue(result); boolean exceptionReceived = false; Future<Boolean> result2 = null; TestSemaphoreCommandWithSlowFallback command2 = null; TestSemaphoreCommandWithSlowFallback command3 = null; try { System.out.println("c2 start: " + System.currentTimeMillis()); command2 = new TestSemaphoreCommandWithSlowFallback(circuitBreaker, 1, 800); result2 = command2.queue(); System.out.println("c2 after queue: " + System.currentTimeMillis()); Thread.sleep(50); System.out.println("c3 start: " + System.currentTimeMillis()); command3 = new TestSemaphoreCommandWithSlowFallback(circuitBreaker, 1, 200); Future<Boolean> result3 = command3.queue(); System.out.println("c3 after queue: " + System.currentTimeMillis()); result3.get(); } catch (Exception e) { e.printStackTrace(); exceptionReceived = true; } assertTrue(result2.get()); if (!exceptionReceived) { fail("We expected an exception on the 2nd get"); } assertCommandExecutionEvents(command1, HystrixEventType.FAILURE, HystrixEventType.FALLBACK_SUCCESS); assertCommandExecutionEvents(command2, HystrixEventType.FAILURE, HystrixEventType.FALLBACK_SUCCESS); assertCommandExecutionEvents(command3, HystrixEventType.FAILURE, HystrixEventType.FALLBACK_REJECTION); assertEquals(0, circuitBreaker.metrics.getCurrentConcurrentExecutionCount()); assertSaneHystrixRequestLog(3); }
@Test public void testExecutionSemaphoreWithQueue() throws Exception { final TestCircuitBreaker circuitBreaker = new TestCircuitBreaker(); TestSemaphoreCommand command1 = new TestSemaphoreCommand(circuitBreaker, 1, 200, TestSemaphoreCommand.RESULT_SUCCESS, TestSemaphoreCommand.FALLBACK_NOT_IMPLEMENTED); boolean result = command1.queue().get(); assertTrue(result); final AtomicBoolean exceptionReceived = new AtomicBoolean(); final TryableSemaphore semaphore = new TryableSemaphoreActual(HystrixProperty.Factory.asProperty(1)); final TestSemaphoreCommand command2 = new TestSemaphoreCommand(circuitBreaker, semaphore, 200, TestSemaphoreCommand.RESULT_SUCCESS, TestSemaphoreCommand.FALLBACK_NOT_IMPLEMENTED); Runnable r2 = new HystrixContextRunnable(HystrixPlugins.getInstance().getConcurrencyStrategy(), new Runnable() { @Override public void run() { try { command2.queue().get(); } catch (Exception e) { e.printStackTrace(); exceptionReceived.set(true); } } }); final TestSemaphoreCommand command3 = new TestSemaphoreCommand(circuitBreaker, semaphore, 200, TestSemaphoreCommand.RESULT_SUCCESS, TestSemaphoreCommand.FALLBACK_NOT_IMPLEMENTED); Runnable r3 = new HystrixContextRunnable(HystrixPlugins.getInstance().getConcurrencyStrategy(), new Runnable() { @Override public void run() { try { command3.queue().get(); } catch (Exception e) { e.printStackTrace(); exceptionReceived.set(true); } } }); Thread t2 = new Thread(r2); Thread t3 = new Thread(r3); t2.start(); Thread.sleep(50); t3.start(); t2.join(); t3.join(); if (!exceptionReceived.get()) { fail("We expected an exception on the 2nd get"); } assertCommandExecutionEvents(command1, HystrixEventType.SUCCESS); assertCommandExecutionEvents(command2, HystrixEventType.SUCCESS); assertCommandExecutionEvents(command3, HystrixEventType.SEMAPHORE_REJECTED, HystrixEventType.FALLBACK_MISSING); assertEquals(0, circuitBreaker.metrics.getCurrentConcurrentExecutionCount()); assertSaneHystrixRequestLog(3); }
@Test public void testRequestCache1() throws Exception { TestCircuitBreaker circuitBreaker = new TestCircuitBreaker(); SuccessfulCacheableCommand<String> command1 = new SuccessfulCacheableCommand<String>(circuitBreaker, true, "A"); SuccessfulCacheableCommand<String> command2 = new SuccessfulCacheableCommand<String>(circuitBreaker, true, "A"); assertTrue(command1.isCommandRunningInThread()); Future<String> f1 = command1.queue(); Future<String> f2 = command2.queue(); assertEquals("A", f1.get()); assertEquals("A", f2.get()); assertTrue(command1.executed); assertFalse(command2.executed); assertTrue(command1.getExecutionTimeInMilliseconds() > -1); assertFalse(command1.isResponseFromCache()); assertTrue(command2.isResponseFromCache()); assertCommandExecutionEvents(command1, HystrixEventType.SUCCESS); assertCommandExecutionEvents(command2, HystrixEventType.SUCCESS, HystrixEventType.RESPONSE_FROM_CACHE); assertEquals(0, circuitBreaker.metrics.getCurrentConcurrentExecutionCount()); assertSaneHystrixRequestLog(2); }
@Test public void testRequestCache2() throws Exception { TestCircuitBreaker circuitBreaker = new TestCircuitBreaker(); SuccessfulCacheableCommand<String> command1 = new SuccessfulCacheableCommand<String>(circuitBreaker, true, "A"); SuccessfulCacheableCommand<String> command2 = new SuccessfulCacheableCommand<String>(circuitBreaker, true, "B"); assertTrue(command1.isCommandRunningInThread()); Future<String> f1 = command1.queue(); Future<String> f2 = command2.queue(); assertEquals("A", f1.get()); assertEquals("B", f2.get()); assertTrue(command1.executed); assertTrue(command2.executed); assertTrue(command2.getExecutionTimeInMilliseconds() > -1); assertFalse(command2.isResponseFromCache()); assertCommandExecutionEvents(command1, HystrixEventType.SUCCESS); assertCommandExecutionEvents(command2, HystrixEventType.SUCCESS); assertNull(command1.getExecutionException()); assertFalse(command2.isResponseFromCache()); assertNull(command2.getExecutionException()); assertEquals(0, circuitBreaker.metrics.getCurrentConcurrentExecutionCount()); assertSaneHystrixRequestLog(2); }
@Test public void testRequestCache3() throws Exception { TestCircuitBreaker circuitBreaker = new TestCircuitBreaker(); SuccessfulCacheableCommand<String> command1 = new SuccessfulCacheableCommand<String>(circuitBreaker, true, "A"); SuccessfulCacheableCommand<String> command2 = new SuccessfulCacheableCommand<String>(circuitBreaker, true, "B"); SuccessfulCacheableCommand<String> command3 = new SuccessfulCacheableCommand<String>(circuitBreaker, true, "A"); assertTrue(command1.isCommandRunningInThread()); Future<String> f1 = command1.queue(); Future<String> f2 = command2.queue(); Future<String> f3 = command3.queue(); assertEquals("A", f1.get()); assertEquals("B", f2.get()); assertEquals("A", f3.get()); assertTrue(command1.executed); assertTrue(command2.executed); assertFalse(command3.executed); assertTrue(command3.isResponseFromCache()); assertCommandExecutionEvents(command1, HystrixEventType.SUCCESS); assertCommandExecutionEvents(command2, HystrixEventType.SUCCESS); assertCommandExecutionEvents(command3, HystrixEventType.SUCCESS, HystrixEventType.RESPONSE_FROM_CACHE); assertEquals(0, circuitBreaker.metrics.getCurrentConcurrentExecutionCount()); System.out.println("ReqLog : " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString()); assertSaneHystrixRequestLog(3); }
@Test public void testRequestCacheWithSlowExecution() throws Exception { TestCircuitBreaker circuitBreaker = new TestCircuitBreaker(); SlowCacheableCommand command1 = new SlowCacheableCommand(circuitBreaker, "A", 200); SlowCacheableCommand command2 = new SlowCacheableCommand(circuitBreaker, "A", 100); SlowCacheableCommand command3 = new SlowCacheableCommand(circuitBreaker, "A", 100); SlowCacheableCommand command4 = new SlowCacheableCommand(circuitBreaker, "A", 100); Future<String> f1 = command1.queue(); Future<String> f2 = command2.queue(); Future<String> f3 = command3.queue(); Future<String> f4 = command4.queue(); assertEquals("A", f2.get()); assertEquals("A", f3.get()); assertEquals("A", f4.get()); assertEquals("A", f1.get()); assertTrue(command1.executed); assertFalse(command2.executed); assertFalse(command3.executed); assertFalse(command4.executed); assertTrue(command1.getExecutionTimeInMilliseconds() > -1); assertFalse(command1.isResponseFromCache()); assertTrue(command2.getExecutionTimeInMilliseconds() == -1); assertTrue(command2.isResponseFromCache()); assertTrue(command3.isResponseFromCache()); assertTrue(command3.getExecutionTimeInMilliseconds() == -1); assertTrue(command4.isResponseFromCache()); assertTrue(command4.getExecutionTimeInMilliseconds() == -1); assertCommandExecutionEvents(command1, HystrixEventType.SUCCESS); assertCommandExecutionEvents(command2, HystrixEventType.SUCCESS, HystrixEventType.RESPONSE_FROM_CACHE); assertCommandExecutionEvents(command3, HystrixEventType.SUCCESS, HystrixEventType.RESPONSE_FROM_CACHE); assertCommandExecutionEvents(command4, HystrixEventType.SUCCESS, HystrixEventType.RESPONSE_FROM_CACHE); assertEquals(0, circuitBreaker.metrics.getCurrentConcurrentExecutionCount()); assertSaneHystrixRequestLog(4); System.out.println("HystrixRequestLog: " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString()); }
@Test public void testNoRequestCache3() throws Exception { TestCircuitBreaker circuitBreaker = new TestCircuitBreaker(); SuccessfulCacheableCommand<String> command1 = new SuccessfulCacheableCommand<String>(circuitBreaker, false, "A"); SuccessfulCacheableCommand<String> command2 = new SuccessfulCacheableCommand<String>(circuitBreaker, false, "B"); SuccessfulCacheableCommand<String> command3 = new SuccessfulCacheableCommand<String>(circuitBreaker, false, "A"); assertTrue(command1.isCommandRunningInThread()); Future<String> f1 = command1.queue(); Future<String> f2 = command2.queue(); Future<String> f3 = command3.queue(); assertEquals("A", f1.get()); assertEquals("B", f2.get()); assertEquals("A", f3.get()); assertTrue(command1.executed); assertTrue(command2.executed); assertTrue(command3.executed); assertCommandExecutionEvents(command1, HystrixEventType.SUCCESS); assertCommandExecutionEvents(command2, HystrixEventType.SUCCESS); assertCommandExecutionEvents(command3, HystrixEventType.SUCCESS); assertEquals(0, circuitBreaker.metrics.getCurrentConcurrentExecutionCount()); assertSaneHystrixRequestLog(3); }
@Test public void testRequestCacheViaQueueSemaphore1() throws Exception { TestCircuitBreaker circuitBreaker = new TestCircuitBreaker(); SuccessfulCacheableCommandViaSemaphore command1 = new SuccessfulCacheableCommandViaSemaphore(circuitBreaker, true, "A"); SuccessfulCacheableCommandViaSemaphore command2 = new SuccessfulCacheableCommandViaSemaphore(circuitBreaker, true, "B"); SuccessfulCacheableCommandViaSemaphore command3 = new SuccessfulCacheableCommandViaSemaphore(circuitBreaker, true, "A"); assertFalse(command1.isCommandRunningInThread()); Future<String> f1 = command1.queue(); Future<String> f2 = command2.queue(); Future<String> f3 = command3.queue(); assertEquals("A", f1.get()); assertEquals("B", f2.get()); assertEquals("A", f3.get()); assertTrue(command1.executed); assertTrue(command2.executed); assertFalse(command3.executed); assertTrue(command3.isResponseFromCache()); assertTrue(command3.getExecutionTimeInMilliseconds() == -1); assertCommandExecutionEvents(command1, HystrixEventType.SUCCESS); assertCommandExecutionEvents(command2, HystrixEventType.SUCCESS); assertCommandExecutionEvents(command3, HystrixEventType.SUCCESS, HystrixEventType.RESPONSE_FROM_CACHE); assertEquals(0, circuitBreaker.metrics.getCurrentConcurrentExecutionCount()); assertSaneHystrixRequestLog(3); }
@Test public void testNoRequestCacheViaQueueSemaphore1() throws Exception { TestCircuitBreaker circuitBreaker = new TestCircuitBreaker(); SuccessfulCacheableCommandViaSemaphore command1 = new SuccessfulCacheableCommandViaSemaphore(circuitBreaker, false, "A"); SuccessfulCacheableCommandViaSemaphore command2 = new SuccessfulCacheableCommandViaSemaphore(circuitBreaker, false, "B"); SuccessfulCacheableCommandViaSemaphore command3 = new SuccessfulCacheableCommandViaSemaphore(circuitBreaker, false, "A"); assertFalse(command1.isCommandRunningInThread()); Future<String> f1 = command1.queue(); Future<String> f2 = command2.queue(); Future<String> f3 = command3.queue(); assertEquals("A", f1.get()); assertEquals("B", f2.get()); assertEquals("A", f3.get()); assertTrue(command1.executed); assertTrue(command2.executed); assertTrue(command3.executed); assertCommandExecutionEvents(command1, HystrixEventType.SUCCESS); assertCommandExecutionEvents(command2, HystrixEventType.SUCCESS); assertCommandExecutionEvents(command3, HystrixEventType.SUCCESS); assertEquals(0, circuitBreaker.metrics.getCurrentConcurrentExecutionCount()); assertSaneHystrixRequestLog(3); }
@Test public void testBadRequestExceptionViaQueueInThread() throws Exception { TestCircuitBreaker circuitBreaker = new TestCircuitBreaker(); BadRequestCommand command1 = null; try { command1 = new BadRequestCommand(circuitBreaker, ExecutionIsolationStrategy.THREAD); command1.queue().get(); fail("we expect to receive a " + HystrixBadRequestException.class.getSimpleName()); } catch (ExecutionException e) { e.printStackTrace(); if (e.getCause() instanceof HystrixBadRequestException) { } else { fail("We expect a " + HystrixBadRequestException.class.getSimpleName() + " but got a " + e.getClass().getSimpleName()); } } assertCommandExecutionEvents(command1, HystrixEventType.BAD_REQUEST); assertNotNull(command1.getExecutionException()); assertEquals(0, circuitBreaker.metrics.getCurrentConcurrentExecutionCount()); assertSaneHystrixRequestLog(1); }
@Test public void testNonBlockingCommandQueueFiresTimeout() throws Exception { final TestHystrixCommand<Integer> cmd = getCommand(ExecutionIsolationStrategy.THREAD, AbstractTestHystrixCommand.ExecutionResult.SUCCESS, 200, AbstractTestHystrixCommand.FallbackResult.SUCCESS, 50); new Thread() { @Override public void run() { cmd.queue(); } }.start(); Thread.sleep(200); assertTrue(cmd.isExecutionComplete()); assertTrue(cmd.isResponseTimedOut()); assertEquals(0, cmd.metrics.getCurrentConcurrentExecutionCount()); }
@Test public void testInterruptFutureOnTimeout() throws InterruptedException, ExecutionException { InterruptibleCommand cmd = new InterruptibleCommand(new TestCircuitBreaker(), true); Future<Boolean> f = cmd.queue(); Thread.sleep(500); assertTrue(cmd.hasBeenInterrupted()); }
@Test public void testDoNotInterruptFutureOnTimeoutIfPropertySaysNotTo() throws InterruptedException, ExecutionException { InterruptibleCommand cmd = new InterruptibleCommand(new TestCircuitBreaker(), false); Future<Boolean> f = cmd.queue(); Thread.sleep(500); assertFalse(cmd.hasBeenInterrupted()); }
@Test public void testCancelFutureWithInterruptionWhenPropertySaysNotTo() throws InterruptedException, ExecutionException { InterruptibleCommand cmd = new InterruptibleCommand(new TestCircuitBreaker(), true, false, 1000); Future<Boolean> f = cmd.queue(); Thread.sleep(500); f.cancel(true); Thread.sleep(500); try { f.get(); fail("Should have thrown a CancellationException"); } catch (CancellationException e) { assertFalse(cmd.hasBeenInterrupted()); } }
@Test public void testCancelFutureWithInterruption() throws InterruptedException, ExecutionException { InterruptibleCommand cmd = new InterruptibleCommand(new TestCircuitBreaker(), true, true, 1000); Future<Boolean> f = cmd.queue(); Thread.sleep(500); f.cancel(true); Thread.sleep(500); try { f.get(); fail("Should have thrown a CancellationException"); } catch (CancellationException e) { assertTrue(cmd.hasBeenInterrupted()); } }
@Test public void testCancelFutureWithoutInterruption() throws InterruptedException, ExecutionException, TimeoutException { InterruptibleCommand cmd = new InterruptibleCommand(new TestCircuitBreaker(), true, true, 1000); Future<Boolean> f = cmd.queue(); Thread.sleep(500); f.cancel(false); Thread.sleep(500); try { f.get(); fail("Should have thrown a CancellationException"); } catch (CancellationException e) { assertFalse(cmd.hasBeenInterrupted()); } }
@Test public void testCancelledTasksInQueueGetRemoved() throws Exception { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("Cancellation-A"); TestCircuitBreaker circuitBreaker = new TestCircuitBreaker(); SingleThreadedPoolWithQueue pool = new SingleThreadedPoolWithQueue(10, 1); TestCommandRejection command1 = new TestCommandRejection(key, circuitBreaker, pool, 500, 600, TestCommandRejection.FALLBACK_NOT_IMPLEMENTED); TestCommandRejection command2 = new TestCommandRejection(key, circuitBreaker, pool, 500, 600, TestCommandRejection.FALLBACK_NOT_IMPLEMENTED); Future<Boolean> poolFiller = command1.queue(); Observable<Boolean> cmdInQueue = command2.observe(); Subscription s = cmdInQueue.subscribe(); assertEquals(1, pool.queue.size()); s.unsubscribe(); assertEquals(0, pool.queue.size()); poolFiller.get(); assertCommandExecutionEvents(command1, HystrixEventType.SUCCESS); assertCommandExecutionEvents(command2, HystrixEventType.CANCELLED); assertEquals(0, circuitBreaker.metrics.getCurrentConcurrentExecutionCount()); System.out.println("ReqLog : " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString()); assertSaneHystrixRequestLog(2); }