@Test
public void testRequestThenCacheHitAndOriginalUnsubscribed() throws Exception {

    TestCollapserTimer timer = new TestCollapserTimer();
    HystrixCollapser<List<String>, String, String> collapser1 = new SuccessfulCacheableCollapsedCommand(timer, "foo", true);
    Observable<String> response1 = collapser1.observe();
    HystrixCollapser<List<String>, String, String> collapser2 = new SuccessfulCacheableCollapsedCommand(timer, "foo", true);
    Observable<String> response2 = collapser2.observe();
    final CountDownLatch latch1 = new CountDownLatch(1);
    final CountDownLatch latch2 = new CountDownLatch(1);
    final AtomicReference<String> value1 = new AtomicReference<String>(null);
    final AtomicReference<String> value2 = new AtomicReference<String>(null);
    Subscription s1 = response1
            .doOnUnsubscribe(new Action0() {
                @Override
                public void call() {
                    System.out.println(System.currentTimeMillis() + " : s1 Unsubscribed!");
                    latch1.countDown();
                }
            })
            .subscribe(new Subscriber<String>() {
                @Override
                public void onCompleted() {
                    System.out.println(System.currentTimeMillis() + " : s1 OnCompleted");
                    latch1.countDown();
                }
                @Override
                public void onError(Throwable e) {
                    System.out.println(System.currentTimeMillis() + " : s1 OnError : " + e);
                    latch1.countDown();
                }
                @Override
                public void onNext(String s) {
                    System.out.println(System.currentTimeMillis() + " : s1 OnNext : " + s);
                    value1.set(s);
                }
            });
    Subscription s2 = response2
            .doOnUnsubscribe(new Action0() {
                @Override
                public void call() {
                    System.out.println(System.currentTimeMillis() + " : s2 Unsubscribed!");
                    latch2.countDown();
                }
            })
            .subscribe(new Subscriber<String>() {
                @Override
                public void onCompleted() {
                    System.out.println(System.currentTimeMillis() + " : s2 OnCompleted");
                    latch2.countDown();
                }
                @Override
                public void onError(Throwable e) {
                    System.out.println(System.currentTimeMillis() + " : s2 OnError : " + e);
                    latch2.countDown();
                }
                @Override
                public void onNext(String s) {
                    System.out.println(System.currentTimeMillis() + " : s2 OnNext : " + s);
                    value2.set(s);
                }
            });
    s1.unsubscribe();
    timer.incrementTime(10); // let time pass that equals the default delay/period
    assertTrue(latch1.await(1000, TimeUnit.MILLISECONDS));
    assertTrue(latch2.await(1000, TimeUnit.MILLISECONDS));
    assertNull(value1.get());
    assertEquals("foo", value2.get());
    System.out.println("ReqLog : " + HystrixRequestLog.getCurrentRequest().getExecutedCommandsAsString());
    assertEquals(1, HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().size());
    HystrixInvokableInfo<?> command = HystrixRequestLog.getCurrentRequest().getAllExecutedCommands().iterator().next();
    assertCommandExecutionEvents(command, HystrixEventType.SUCCESS, HystrixEventType.COLLAPSED);
    assertEquals(1, command.getNumberCollapsed()); //should only be 1 collapsed - other came from cache, then was cancelled
}