Behaviour of doFinally

[LukasHavemann]

Expected behavior

@Test
public void errorBeforeFinallyTest() {
    System.err.println("> error before finally");
    Mono.just(true)
        .map(this::throwError)
        .doFinally(this::doFinalThing)
        .subscribe();
}

@Test
public void errorAfterFinallyTest() {
    System.err.println("> error after finally");
    Mono.just(true)
        .doFinally(this::doFinalThing)
        .map(this::throwError)
        .subscribe();
}

private void doFinalThing(SignalType any) {
    System.err.println("do finally called");
}

private Boolean throwError(Boolean x) {
    System.err.println("throw exception");
    throw new IllegalStateException();
}

According to the javadoc shouldn't be both tests have the same log output?

Actual behavior

> error after finally
throw exception
do finally called


> error before finally
throw exception

Steps to reproduce

See code above

Reactor Core version

3.1.1

JVM version (e.g. java -version)

java version "1.8.0_151"
Java(TM) SE Runtime Environment (build 1.8.0_151-b12)
Java HotSpot(TM) 64-Bit Server VM (build 25.151-b12, mixed mode)

OS version (e.g. uname -a)

MINGW64_NT-10.0

[simonbasle]

That's not what I see, the second example goes through doFinalThing as well, despite the error being downstream of doFinally. That's because the error in the map will trigger a cancellation upstream, which is seen by doFinally: you can make that obvious by appending the any variable to the printout.

Modifying doFinalThing accordingly and slightly amending your code to have the subscribe swallow the exception, I have made a quick test that performs both tasks:

@Test
public void errorBeforeAndAfterFinallyTest() {
	System.err.println("> error before finally");
	Mono.just(true)
	    .map(this::throwError)
	    .doFinally(this::doFinalThing)
	    .subscribe(v -> {}, e -> {});

	System.err.println("> error after finally");
	Mono.just(true)
	    .doFinally(this::doFinalThing)
	    .map(this::throwError)
	    .subscribe(v -> {}, e -> {});
}

I get the following output:

> error before finally
throw exception
do finally called with onError
> error after finally
throw exception
do finally called with cancel

[LukasHavemann]

Thank you for the clarification!

[baptistemesta]

Hi, I'm commenting in that because I think there is still an issue on the doFinally.

    @Test
    public void withoutConsumerInSubscribe() {
        System.err.println("> subscribe without consumer: map_doOnError_doFinally");
        Mono.just(true)
                .map(this::throwError)
                .doOnError(e -> System.err.println("doOnError"))
                .doFinally(any -> System.err.println("doFinally " + any.toString()))
                .subscribe();
        System.err.println("> subscribe without consumer: doFinally_map_doOnError");
        Mono.just(true)
                .doFinally(any -> System.err.println("doFinally " + any.toString()))
                .map(this::throwError)
                .doOnError(e -> System.err.println("doOnError"))
                .subscribe();
        System.err.println("> subscribe without consumer:  map_doFinally_doOnError");
        Mono.just(true)
                .map(this::throwError)
                .doFinally(any -> System.err.println("doFinally " + any.toString()))
                .doOnError(e -> System.err.println("doOnError"))
                .subscribe();
    }

    @Test
    public void withConsumerInSubscribe() {
        System.err.println("> subscribe with consumer: map_doOnError_doFinally");
        Mono.just(true)
                .map(this::throwError)
                .doOnError(e -> System.err.println("doOnError"))
                .doFinally(any -> System.err.println("doFinally " + any.toString()))
                .subscribe(v -> { }, e -> { });
        System.err.println("> subscribe with consumer: doFinally_map_doOnError");
        Mono.just(true)
                .doFinally(any -> System.err.println("doFinally " + any.toString()))
                .map(this::throwError)
                .doOnError(e -> System.err.println("doOnError"))
                .subscribe(v -> { }, e -> { });
        System.err.println("> subscribe with consumer: map_doFinally_doOnError");
        Mono.just(true)
                .map(this::throwError)
                .doFinally(any -> System.err.println("doFinally " + any.toString()))
                .doOnError(e -> System.err.println("doOnError"))
                .subscribe(v -> { }, e -> { });
    }

    @Test
    public void whithoutDoOnError() {
        System.err.println("> whithoutDoOnError");
        Mono.just(true)
                .map(this::throwError)
                .doFinally(any -> System.err.println("doFinally " + any.toString()))
                .subscribe();
    }

    private Boolean throwError(Boolean x) {
        throw new IllegalStateException();
    }

the produce

> subscribe without consumer: map_doOnError_doFinally
doOnError
> subscribe without consumer: doFinally_map_doOnError
doFinally cancel
doOnError
> subscribe without consumer:  map_doFinally_doOnError
doOnError
doFinally onError
> subscribe with consumer: map_doOnError_doFinally
doOnError
doFinally onError
> subscribe with consumer: doFinally_map_doOnError
doFinally cancel
doOnError
> subscribe with consumer: map_doFinally_doOnError
doOnError
doFinally onError

> whithoutDoOnError
TestFinally > whithoutDoOnError FAILED
    reactor.core.Exceptions$ErrorCallbackNotImplemented
        Caused by: java.lang.IllegalStateException at TestFinally.java:69

java.lang.IllegalStateException
reactor.core.Exceptions$ErrorCallbackNotImplemented: java.lang.IllegalStateException
Caused by: java.lang.IllegalStateException
	at TestFinally.throwError(TestFinally.java:69)
[...]

As you can see, when using .subscribe() instead of .subscribe(consumer, errorConsumer) the doFinally is not executed if it is placed after the doOnError.

Also, I upgraded from 3.0.7 to 3.1.2 and its not possible to write the code from whithoutDoOnError with the subscribe() without parameters.

Reactor Core version

3.1.2

[simonbasle]

Thanks for the demonstration @baptistemesta. Not sure what you mean by "it is not possible to write..." (copy-pasted the whole code in a test and it worked fine from a compilation point of view).

Indeed the case where no error handler is provided in subscribe() and the doFinally is closest to the subscribe will skip the handler.

This is due to the subscribe throwing an UnsupportedOperationException complaining about the missing error handler, and the doFinally not being protected against that. In other cases, the doOnError does protect against said UOE, so in effect the doFinally is unaffected.

[baptistemesta]

That was fast :)
thank you!

[simonbasle]

well the second repro case was a huge help, so thanks for providing that ;) 👍

[baptistemesta]

BTW when I was saying it is not possible to write I meant the code was not throwing exception on whithoutDoOnError i.e.. the UnsupportedOperationException

[unoexperto]

I just spent 3 hrs trying to debug why I'm getting SignalType.CANCEL on exception thrown upstream. I wish this behavior was cleaner. It's very very confusing right now. I still don't understand why it's happening. Could you please explain it?

Schematically my code looks like this:

Mono.fromCallable {
    error("zzz")
}
    .doFinally { signal ->
       // get CANCEL here
    }
    .subscribe({ event ->
        mainSink.emitNext(event, busyLooping)
    }, { ex ->
        // error
        flushingSink.emitError(ex, busyLooping)
    }, {
        // complete
        flushingSink.emitComplete(busyLooping)
    })

[chemicL]

I just spent 3 hrs trying to debug why I'm getting SignalType.CANCEL on exception thrown upstream. I wish this behavior was cleaner. It's very very confusing right now. I still don't understand why it's happening. Could you please explain it?

Schematically my code looks like this:

Mono.fromCallable {
    error("zzz")
}
    .doFinally { signal ->
       // get CANCEL here
    }
    .subscribe({ event ->
        mainSink.emitNext(event, busyLooping)
    }, { ex ->
        // error
        flushingSink.emitError(ex, busyLooping)
    }, {
        // complete
        flushingSink.emitComplete(busyLooping)
    })

@unoexperto Please provide a complete reproducer (that we can run locally) either as an issue if you think you're facing a bug, or as a question on stackoverflow (https://stackoverflow.com/questions/tagged/project-reactor).