@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");
        Flowable<String> w = Flowable.unsafeCreate(onSubscribe);
        BusySubscriber busySubscriber = new BusySubscriber();
        Subscriber<String> aw = serializedSubscriber(busySubscriber);
        w.subscribe(aw);
        onSubscribe.waitToFinish();
        System.out.println("OnSubscribe maxConcurrentThreads: " + onSubscribe.maxConcurrentThreads.get() + "  Subscriber maxConcurrentThreads: " + busySubscriber.maxConcurrentThreads.get());
        // we can have concurrency ...
        assertTrue(onSubscribe.maxConcurrentThreads.get() > 1);
        // ... but the onNext execution should be single threaded
        assertEquals(1, busySubscriber.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: " + busySubscriber.onNextCount.get());
        assertFalse(busySubscriber.onComplete);
        assertTrue(busySubscriber.onError);
        assertTrue(busySubscriber.onNextCount.get() < 9);
        // no onComplete 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();
    }
}