@Test
public void testDuplicateArgumentsWithRequestCachingOff() throws Exception {

    final int NUM = 10;
    List<Observable<Integer>> observables = new ArrayList<Observable<Integer>>();
    for (int i = 0; i < NUM; i++) {
        MyCollapser c = new MyCollapser("5", false);
        observables.add(c.toObservable());
    }
    List<TestSubscriber<Integer>> subscribers = new ArrayList<TestSubscriber<Integer>>();
    for (final Observable<Integer> o: observables) {
        final TestSubscriber<Integer> sub = new TestSubscriber<Integer>();
        subscribers.add(sub);
        o.subscribe(sub);
    }
    Thread.sleep(100);
    AtomicInteger numErrors = new AtomicInteger(0);
    AtomicInteger numValues = new AtomicInteger(0);
    // only the first subscriber should receive the value.
    // the others should get an error that the batch contains duplicates
    for (TestSubscriber<Integer> sub: subscribers) {
        sub.awaitTerminalEvent(1000, TimeUnit.MILLISECONDS);
        if (sub.getOnCompletedEvents().isEmpty()) {
            System.out.println(Thread.currentThread().getName() + " Error : " + sub.getOnErrorEvents());
            sub.assertError(IllegalArgumentException.class);
            sub.assertNoValues();
            numErrors.getAndIncrement();
        } else {
            System.out.println(Thread.currentThread().getName() + " OnNext : " + sub.getOnNextEvents());
            sub.assertValues(5);
            sub.assertCompleted();
            sub.assertNoErrors();
            numValues.getAndIncrement();
        }
    }
    assertEquals(1, numValues.get());
    assertEquals(NUM - 1, numErrors.get());
}