Not entering error handling block during exception

Question

In the following method, I am purposely making the call c.rxCommit() throw an exception during test. The exception is thrown as expected, but I never land in the onErrorResumeNext block as expected.

I am expecting to land inside here to handle the error and perform some rollback. Can I get some advice as to why I am not landing within the error block? Thanks.

public Observable<Void> myMethod(Observable<Map<String, String>> records) {

    return client.rxGetConnection()
            .flatMapCompletable(c -> c.rxSetAutoCommit(false)
                    .toCompletable()
                    .andThen(records.flatMapSingle(record -> perform(c, record)).toCompletable())
                    .andThen(c.rxCommit().toCompletable()) // test makes this c.rxCommit() throw an error on purpose. 
                    .onErrorResumeNext(throwable -> {
                        // I want to land in here when error but it is not happening.
                        return c.rxRollback().toCompletable();
                    })
                    .andThen(c.rxSetAutoCommit(true).toCompletable())
                    .andThen(c.rxClose()).toCompletable()
            ).toObservable();
}

The test

@Test
void test() {
    when(sqlConnection.rxCommit()).thenThrow(new RuntimeException("Error on Commit")); // mockito
    myClass.myMethod(mockedRecords).subscribe(
            success -> System.out.println(),
            throwable -> System.out.println("err " + throwable), // I land in here with the new RuntimeException("Error on Commit") exception. 
            // this is wrong. Error should have been handled in the onErrorResumeNext block and should be getting success here instead. 
    );

    // some verify() to test outcome
}

Answer 1

As akarnokd says, your sqlConnection.rxCommit() mock is wrong, because it throws exception right after method call, not by subscription.

If want to receive error in rx flow, try this:

when(sqlConnection.rxCommit()).thenReturn(Single.error(RuntimeException("Error on Commit")));

But if you really want to throw exceptions in rxCommit(), try to wrap it in Single.defer:

.andThen(
    Single.defer(() -> {
        return c.rxCommit().toCompletable();
})