RollingThreadPoolMaxConcurrencyStream extends RollingConcurrencyStream { public static RollingThreadPoolMaxConcurrencyStream getInstance(HystrixThreadPoolKey threadPoolKey, HystrixThreadPoolProperties properties) { final int counterMetricWindow = properties.metricsRollingStatisticalWindowInMilliseconds().get(); final int numCounterBuckets = properties.metricsRollingStatisticalWindowBuckets().get(); final int counterBucketSizeInMs = counterMetricWindow / numCounterBuckets; return getInstance(threadPoolKey, numCounterBuckets, counterBucketSizeInMs); } RollingThreadPoolMaxConcurrencyStream(final HystrixThreadPoolKey threadPoolKey, final int numBuckets, final int bucketSizeInMs); static RollingThreadPoolMaxConcurrencyStream getInstance(HystrixThreadPoolKey threadPoolKey, HystrixThreadPoolProperties properties); static RollingThreadPoolMaxConcurrencyStream getInstance(HystrixThreadPoolKey threadPoolKey, int numBuckets, int bucketSizeInMs); static void reset();  }
@Test public void testEmptyStreamProducesZeros() { HystrixCommandGroupKey groupKey = HystrixCommandGroupKey.Factory.asKey("ThreadPool-Concurrency-A"); HystrixThreadPoolKey threadPoolKey = HystrixThreadPoolKey.Factory.asKey("ThreadPool-Concurrency-A"); HystrixCommandKey key = HystrixCommandKey.Factory.asKey("RollingConcurrency-A"); stream = RollingThreadPoolMaxConcurrencyStream.getInstance(threadPoolKey, 10, 100); stream.startCachingStreamValuesIfUnstarted(); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(getSubscriber(latch)); try { assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); } catch (InterruptedException ex) { fail("Interrupted ex"); } assertEquals(0, stream.getLatestRollingMax()); }
@Test public void testStartsAndEndsInSameBucketProduceValue() throws InterruptedException { HystrixCommandGroupKey groupKey = HystrixCommandGroupKey.Factory.asKey("ThreadPool-Concurrency-B"); HystrixThreadPoolKey threadPoolKey = HystrixThreadPoolKey.Factory.asKey("ThreadPool-Concurrency-B"); HystrixCommandKey key = HystrixCommandKey.Factory.asKey("RollingConcurrency-B"); stream = RollingThreadPoolMaxConcurrencyStream.getInstance(threadPoolKey, 10, 100); stream.startCachingStreamValuesIfUnstarted(); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(getSubscriber(latch)); Command cmd1 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 50); Command cmd2 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 40); cmd1.observe(); Thread.sleep(1); cmd2.observe(); assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); assertEquals(2, stream.getLatestRollingMax()); }
@Test public void testStartsAndEndsInSameBucketSemaphoreIsolated() throws InterruptedException { HystrixCommandGroupKey groupKey = HystrixCommandGroupKey.Factory.asKey("ThreadPool-Concurrency-C"); HystrixThreadPoolKey threadPoolKey = HystrixThreadPoolKey.Factory.asKey("ThreadPool-Concurrency-C"); HystrixCommandKey key = HystrixCommandKey.Factory.asKey("RollingConcurrency-C"); stream = RollingThreadPoolMaxConcurrencyStream.getInstance(threadPoolKey, 10, 100); stream.startCachingStreamValuesIfUnstarted(); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(getSubscriber(latch)); Command cmd1 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 10, HystrixCommandProperties.ExecutionIsolationStrategy.SEMAPHORE); Command cmd2 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 14, HystrixCommandProperties.ExecutionIsolationStrategy.SEMAPHORE); cmd1.observe(); Thread.sleep(1); cmd2.observe(); assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); assertEquals(0, stream.getLatestRollingMax()); }
@Test public void testOneCommandCarriesOverToNextBucket() throws InterruptedException { HystrixCommandGroupKey groupKey = HystrixCommandGroupKey.Factory.asKey("ThreadPool-Concurrency-D"); HystrixThreadPoolKey threadPoolKey = HystrixThreadPoolKey.Factory.asKey("ThreadPool-Concurrency-D"); HystrixCommandKey key = HystrixCommandKey.Factory.asKey("RollingConcurrency-D"); stream = RollingThreadPoolMaxConcurrencyStream.getInstance(threadPoolKey, 10, 100); stream.startCachingStreamValuesIfUnstarted(); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(getSubscriber(latch)); Command cmd1 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 560); Command cmd2 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 50); Command cmd3 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 75); cmd1.observe(); Thread.sleep(150); cmd2.observe(); Thread.sleep(1); cmd3.observe(); assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); assertEquals(3, stream.getLatestRollingMax()); }
@Test public void testMultipleCommandsCarryOverMultipleBuckets() throws InterruptedException { HystrixCommandGroupKey groupKey = HystrixCommandGroupKey.Factory.asKey("ThreadPool-Concurrency-E"); HystrixThreadPoolKey threadPoolKey = HystrixThreadPoolKey.Factory.asKey("ThreadPool-Concurrency-E"); HystrixCommandKey key = HystrixCommandKey.Factory.asKey("RollingConcurrency-E"); stream = RollingThreadPoolMaxConcurrencyStream.getInstance(threadPoolKey, 10, 100); stream.startCachingStreamValuesIfUnstarted(); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(getSubscriber(latch)); Command cmd1 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 300); Command cmd2 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 300); Command cmd3 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 10); Command cmd4 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 10); cmd1.observe(); Thread.sleep(100); cmd2.observe(); Thread.sleep(100); cmd3.observe(); Thread.sleep(100); cmd4.observe(); assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); assertEquals(3, stream.getLatestRollingMax()); }
@Test public void testMultipleCommandsCarryOverMultipleBucketsForMultipleThreadPools() throws InterruptedException { HystrixCommandGroupKey groupKeyX = HystrixCommandGroupKey.Factory.asKey("ThreadPool-Concurrency-X"); HystrixCommandGroupKey groupKeyY = HystrixCommandGroupKey.Factory.asKey("ThreadPool-Concurrency-Y"); HystrixThreadPoolKey threadPoolKey = HystrixThreadPoolKey.Factory.asKey("ThreadPool-Concurrency-X"); HystrixCommandKey keyX = HystrixCommandKey.Factory.asKey("RollingConcurrency-X"); HystrixCommandKey keyY = HystrixCommandKey.Factory.asKey("RollingConcurrency-Y"); stream = RollingThreadPoolMaxConcurrencyStream.getInstance(threadPoolKey, 10, 100); stream.startCachingStreamValuesIfUnstarted(); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(getSubscriber(latch)); Command cmd1 = Command.from(groupKeyX, keyX, HystrixEventType.SUCCESS, 300); Command cmd2 = Command.from(groupKeyY, keyY, HystrixEventType.SUCCESS, 300); Command cmd3 = Command.from(groupKeyX, keyY, HystrixEventType.SUCCESS, 10); Command cmd4 = Command.from(groupKeyX, keyY, HystrixEventType.SUCCESS, 10); cmd1.observe(); Thread.sleep(100); cmd2.observe(); Thread.sleep(100); cmd3.observe(); Thread.sleep(100); cmd4.observe(); assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); assertEquals(2, stream.getLatestRollingMax()); }
@Test public void testMultipleCommandsCarryOverMultipleBucketsAndThenAgeOut() throws InterruptedException { HystrixCommandGroupKey groupKey = HystrixCommandGroupKey.Factory.asKey("ThreadPool-Concurrency-F"); HystrixThreadPoolKey threadPoolKey = HystrixThreadPoolKey.Factory.asKey("ThreadPool-Concurrency-F"); HystrixCommandKey key = HystrixCommandKey.Factory.asKey("RollingConcurrency-F"); stream = RollingThreadPoolMaxConcurrencyStream.getInstance(threadPoolKey, 10, 100); stream.startCachingStreamValuesIfUnstarted(); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(30).subscribe(getSubscriber(latch)); Command cmd1 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 300); Command cmd2 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 300); Command cmd3 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 10); Command cmd4 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 10); cmd1.observe(); Thread.sleep(100); cmd2.observe(); Thread.sleep(100); cmd3.observe(); Thread.sleep(100); cmd4.observe(); assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); assertEquals(0, stream.getLatestRollingMax()); }
@Test public void testConcurrencyStreamProperlyFiltersOutResponseFromCache() throws InterruptedException { HystrixCommandGroupKey groupKey = HystrixCommandGroupKey.Factory.asKey("ThreadPool-Concurrency-G"); HystrixThreadPoolKey threadPoolKey = HystrixThreadPoolKey.Factory.asKey("ThreadPool-Concurrency-G"); HystrixCommandKey key = HystrixCommandKey.Factory.asKey("RollingConcurrency-G"); stream = RollingThreadPoolMaxConcurrencyStream.getInstance(threadPoolKey, 10, 100); stream.startCachingStreamValuesIfUnstarted(); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(getSubscriber(latch)); Command cmd1 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 40); Command cmd2 = Command.from(groupKey, key, HystrixEventType.RESPONSE_FROM_CACHE); Command cmd3 = Command.from(groupKey, key, HystrixEventType.RESPONSE_FROM_CACHE); Command cmd4 = Command.from(groupKey, key, HystrixEventType.RESPONSE_FROM_CACHE); cmd1.observe(); Thread.sleep(5); cmd2.observe(); cmd3.observe(); cmd4.observe(); assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); System.out.println("ReqLog : " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString()); assertTrue(cmd2.isResponseFromCache()); assertTrue(cmd3.isResponseFromCache()); assertTrue(cmd4.isResponseFromCache()); assertEquals(1, stream.getLatestRollingMax()); }
@Test public void testConcurrencyStreamProperlyFiltersOutShortCircuits() throws InterruptedException { HystrixCommandGroupKey groupKey = HystrixCommandGroupKey.Factory.asKey("ThreadPool-Concurrency-H"); HystrixThreadPoolKey threadPoolKey = HystrixThreadPoolKey.Factory.asKey("ThreadPool-Concurrency-H"); HystrixCommandKey key = HystrixCommandKey.Factory.asKey("RollingConcurrency-H"); stream = RollingThreadPoolMaxConcurrencyStream.getInstance(threadPoolKey, 10, 100); stream.startCachingStreamValuesIfUnstarted(); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(getSubscriber(latch)); Command failure1 = Command.from(groupKey, key, HystrixEventType.FAILURE); Command failure2 = Command.from(groupKey, key, HystrixEventType.FAILURE); Command failure3 = Command.from(groupKey, key, HystrixEventType.FAILURE); List<Command> shortCircuited = new ArrayList<Command>(); for (int i = 0; i < 20; i++) { shortCircuited.add(Command.from(groupKey, key, HystrixEventType.SUCCESS, 100)); } failure1.execute(); failure2.execute(); failure3.execute(); Thread.sleep(150); for (Command cmd: shortCircuited) { cmd.observe(); } assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); System.out.println("ReqLog : " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString()); for (Command cmd: shortCircuited) { assertTrue(cmd.isResponseShortCircuited()); } assertEquals(1, stream.getLatestRollingMax()); }
@Test public void testConcurrencyStreamProperlyFiltersOutSemaphoreRejections() throws InterruptedException { HystrixCommandGroupKey groupKey = HystrixCommandGroupKey.Factory.asKey("ThreadPool-Concurrency-I"); HystrixThreadPoolKey threadPoolKey = HystrixThreadPoolKey.Factory.asKey("ThreadPool-Concurrency-I"); HystrixCommandKey key = HystrixCommandKey.Factory.asKey("RollingConcurrency-I"); stream = RollingThreadPoolMaxConcurrencyStream.getInstance(threadPoolKey, 10, 100); stream.startCachingStreamValuesIfUnstarted(); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(getSubscriber(latch)); List<Command> saturators = new ArrayList<Command>(); for (int i = 0; i < 10; i++) { saturators.add(Command.from(groupKey, key, HystrixEventType.SUCCESS, 400, HystrixCommandProperties.ExecutionIsolationStrategy.SEMAPHORE)); } final List<Command> rejected = new ArrayList<Command>(); for (int i = 0; i < 10; i++) { rejected.add(Command.from(groupKey, key, HystrixEventType.SUCCESS, 100, HystrixCommandProperties.ExecutionIsolationStrategy.SEMAPHORE)); } for (final Command saturatingCmd: saturators) { threadPool.submit(new HystrixContextRunnable(new Runnable() { @Override public void run() { saturatingCmd.observe(); } })); } Thread.sleep(30); for (final Command rejectedCmd: rejected) { threadPool.submit(new HystrixContextRunnable(new Runnable() { @Override public void run() { rejectedCmd.observe(); } })); } assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); System.out.println("ReqLog : " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString()); for (Command rejectedCmd: rejected) { assertTrue(rejectedCmd.isResponseSemaphoreRejected() || rejectedCmd.isResponseShortCircuited()); } assertEquals(0, stream.getLatestRollingMax()); }
@Test public void testConcurrencyStreamProperlyFiltersOutThreadPoolRejections() throws InterruptedException { HystrixCommandGroupKey groupKey = HystrixCommandGroupKey.Factory.asKey("ThreadPool-Concurrency-J"); HystrixThreadPoolKey threadPoolKey = HystrixThreadPoolKey.Factory.asKey("ThreadPool-Concurrency-J"); HystrixCommandKey key = HystrixCommandKey.Factory.asKey("RollingConcurrency-J"); stream = RollingThreadPoolMaxConcurrencyStream.getInstance(threadPoolKey, 10, 100); stream.startCachingStreamValuesIfUnstarted(); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(10).subscribe(getSubscriber(latch)); List<Command> saturators = new ArrayList<Command>(); for (int i = 0; i < 10; i++) { saturators.add(Command.from(groupKey, key, HystrixEventType.SUCCESS, 400)); } final List<Command> rejected = new ArrayList<Command>(); for (int i = 0; i < 10; i++) { rejected.add(Command.from(groupKey, key, HystrixEventType.SUCCESS, 100)); } for (final Command saturatingCmd: saturators) { saturatingCmd.observe(); } Thread.sleep(30); for (final Command rejectedCmd: rejected) { rejectedCmd.observe(); } assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); System.out.println("ReqLog : " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString()); for (Command rejectedCmd: rejected) { assertTrue(rejectedCmd.isResponseThreadPoolRejected()); } assertEquals(10, stream.getLatestRollingMax()); }