@Test
public void ignoreErrorsFromDownstream() {

    final Instant now = Instant.ofEpochMilli(testScheduler.now());
    final RateLimitedBatcher<BatchableOperationMock, String> batcher = buildBatcher(minimumTimeInQueueMs);
    final Subject<BatchableOperationMock, BatchableOperationMock> updates = PublishSubject.<BatchableOperationMock>create().toSerialized();
    final AssertableSubscriber<?> subscriber = updates.lift(batcher)
            .lift(new ExceptionThrowingOperator("some error happened"))
            .test();
    testScheduler.triggerActions();
    subscriber.assertNoTerminalEvent().assertNoValues();
    for (int i = 0; i < 10; i++) {
        updates.onNext(new BatchableOperationMock(LOW, now, "resource2", "sub2", "create"));
        testScheduler.advanceTimeBy(minimumTimeInQueueMs, TimeUnit.MILLISECONDS);
        subscriber.assertNoTerminalEvent().assertNoValues();
    }
    updates.onCompleted();
    testScheduler.advanceTimeBy(minimumTimeInQueueMs, TimeUnit.MILLISECONDS);
    subscriber.assertNoValues().assertCompleted();
}