@Test
public void testMultiThreadedWithNPEinMiddle() {

    int n = 10;
    for (int i = 0; i < n; i++) {
        TestMultiThreadedObservable onSubscribe = new TestMultiThreadedObservable("one", "two", "three", null, 
                "four", "five", "six", "seven", "eight", "nine");
        Observable<String> w = Observable.create(onSubscribe);
        BusyObserver busyObserver = new BusyObserver();
        Observer<String> aw = serializedObserver(busyObserver);
        w.subscribe(aw);
        onSubscribe.waitToFinish();
        System.out.println("OnSubscribe maxConcurrentThreads: " + onSubscribe.maxConcurrentThreads.get() + "  Observer maxConcurrentThreads: " + busyObserver.maxConcurrentThreads.get());
        // we can have concurrency ...
        assertTrue(onSubscribe.maxConcurrentThreads.get() > 1);
        // ... but the onNext execution should be single threaded
        assertEquals(1, busyObserver.maxConcurrentThreads.get());
        // this should not be the full number of items since the error should stop it before it completes all 9
        System.out.println("onNext count: " + busyObserver.onNextCount.get());
        assertFalse(busyObserver.onCompleted);
        assertTrue(busyObserver.onError);
        assertTrue(busyObserver.onNextCount.get() < 9);
        // no onCompleted because onError was invoked
        // non-deterministic because unsubscribe happens after 'waitToFinish' releases
        // so commenting out for now as this is not a critical thing to test here
        // verify(s, times(1)).unsubscribe();
    }
}