RollingCommandMaxConcurrencyStream extends RollingConcurrencyStream { public static RollingCommandMaxConcurrencyStream getInstance(HystrixCommandKey commandKey, HystrixCommandProperties properties) { final int counterMetricWindow = properties.metricsRollingStatisticalWindowInMilliseconds().get(); final int numCounterBuckets = properties.metricsRollingStatisticalWindowBuckets().get(); final int counterBucketSizeInMs = counterMetricWindow / numCounterBuckets; return getInstance(commandKey, numCounterBuckets, counterBucketSizeInMs); } private  RollingCommandMaxConcurrencyStream(final HystrixCommandKey commandKey, final int numBuckets, final int bucketSizeInMs); static RollingCommandMaxConcurrencyStream getInstance(HystrixCommandKey commandKey, HystrixCommandProperties properties); static RollingCommandMaxConcurrencyStream getInstance(HystrixCommandKey commandKey, int numBuckets, int bucketSizeInMs); static void reset();  }
@Test public void testEmptyStreamProducesZeros() { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Concurrency-A"); stream = RollingCommandMaxConcurrencyStream.getInstance(key, 10, 500); stream.startCachingStreamValuesIfUnstarted(); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(5).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 { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Concurrency-B"); stream = RollingCommandMaxConcurrencyStream.getInstance(key, 10, 500); stream.startCachingStreamValuesIfUnstarted(); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(5).subscribe(getSubscriber(latch)); Command cmd1 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 100); Command cmd2 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 100); cmd1.observe(); Thread.sleep(1); cmd2.observe(); assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); assertEquals(2, stream.getLatestRollingMax()); }
@Test public void testOneCommandCarriesOverToNextBucket() throws InterruptedException { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Concurrency-C"); stream = RollingCommandMaxConcurrencyStream.getInstance(key, 10, 100); stream.startCachingStreamValuesIfUnstarted(); final CountDownLatch latch = new CountDownLatch(1); stream.observe().take(5).subscribe(getSubscriber(latch)); Command cmd1 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 160); Command cmd2 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 10); Command cmd3 = Command.from(groupKey, key, HystrixEventType.SUCCESS, 15); cmd1.observe(); Thread.sleep(100); cmd2.observe(); Thread.sleep(1); cmd3.observe(); assertTrue(latch.await(10000, TimeUnit.MILLISECONDS)); assertEquals(3, stream.getLatestRollingMax()); }
@Test public void testMultipleCommandsCarryOverMultipleBuckets() throws InterruptedException { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Concurrency-D"); stream = RollingCommandMaxConcurrencyStream.getInstance(key, 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 testMultipleCommandsCarryOverMultipleBucketsAndThenAgeOut() throws InterruptedException { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Concurrency-E"); stream = RollingCommandMaxConcurrencyStream.getInstance(key, 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 { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Concurrency-F"); stream = RollingCommandMaxConcurrencyStream.getInstance(key, 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()); assertEquals(1, stream.getLatestRollingMax()); }
@Test public void testConcurrencyStreamProperlyFiltersOutShortCircuits() throws InterruptedException { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Concurrency-G"); stream = RollingCommandMaxConcurrencyStream.getInstance(key, 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()); assertEquals(1, stream.getLatestRollingMax()); }
@Test public void testConcurrencyStreamProperlyFiltersOutSemaphoreRejections() throws InterruptedException { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Concurrency-H"); stream = RollingCommandMaxConcurrencyStream.getInstance(key, 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()); assertEquals(10, stream.getLatestRollingMax()); }
@Test public void testConcurrencyStreamProperlyFiltersOutThreadPoolRejections() throws InterruptedException { HystrixCommandKey key = HystrixCommandKey.Factory.asKey("CMD-Concurrency-I"); stream = RollingCommandMaxConcurrencyStream.getInstance(key, 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()); assertEquals(10, stream.getLatestRollingMax()); }