@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());
}