Skip to main content

[Translation] When Should You Use onErrorContinue()?

tip

Original article: When and how to use onErrorContinue(): Reactor FAQ

The Answer: Never​

I recently received a question about the behavior of the onErrorContinue() operator in Reactor. Honestly, I've never used it in production code.

To be more frank, I didn't fully understand how it works. So I dug into the documentation and some online discussions. In principle, the onErrorContinue() operator should ignore an error and continue running.

So if you have a stream that generates thousands of events and encounters errors on 100 events, you can still continue processing the remaining 900 events. Sounds great, especially compared to onErrorResume().

The latter simply stops the stream and replaces it with another stream. Technically, the replacement stream could be the one that just failed.

This is basically how the retry() operator works: When a stream fails, resubscribe to it

Unfortunately, neither onErrorResume() nor retry() preserves the state of the failed stream. This means retrying might generate the same events we've already processed, or miss some events. It depends on how the initial stream was constructed - simply put, whether it's a hot stream or cold stream.

From this perspective, onErrorContinue() sounds like a good idea - just swallow the errored event and move on! Unfortunately, the onErrorContinue() operator is quite tricky and can lead to some subtle bugs.

Check out this excellent article about onErrorContinue and onErrorResume in Reactor with some interesting examples: Reactor onErrorContinue VS onErrorResume

About onErrorContinue's Design​

I stumbled upon a discussion on GitHub about onErrorContinue() design. In a year-long conversation between confused developers and Reactor library contributors, there's a great quote from one of the founders:

onErrorContinue is a billion dollar mistake I made :(

No need to blame Simon Baslé - designing an API and how it will evolve is very difficult.

Both Reactor and RxJava have historically removed multiple operators. But this quote probably best illustrates how you should handle this operator.

onErrorContinue() promises to skip invalid inputs. Let's use this as an example:

Flux.just("one.txt", "two.txt", "three.txt")
	.flatMap(file -> Mono.fromCallable(() -> new FileInputStream(file)))
	.doOnNext(e -> log.info("Got file {}", e))
	.onErrorContinue(FileNotFoundException.class, (ex, o) -> log.warn("Not found? {}", ex.toString()))
	.onErrorContinue(IOException.class, (ex, o) -> log.warn("I/O error {}", ex.toString()));

When only file two.txt exists, the output is as expected:

WARN - Not found? java.io.FileNotFoundException: one.txt (No such file or directory)
INFO - Got file java.io.FileInputStream@6933b6c6
WARN - Not found? java.io.FileNotFoundException: three.txt (No such file or directory)
warning

I intentionally ignored the exception stack trace. Outside of blog posts, doing this in real projects is absolutely wrong.

Without onErrorContinue(), the stream would fail on the first file. Sounds about right? So what if a slightly modified code snippet throws a more generic IOException instead of FileNotFoundException? Fortunately, we have two onErrorContinue() calls, so we expect to catch IOException in the second one:

Flux
	.just("one.txt", "two.txt", "three.txt")
	.flatMap(file -> Mono.fromCallable(() -> new File("/dev", file).createNewFile()))
	.doOnNext(e -> log.info("Got file {}", e))
	.onErrorContinue(FileNotFoundException.class, (ex, o) -> log.warn("Not found? {}", ex.toString()))
	.onErrorContinue(IOException.class, (ex, o) -> log.warn("I/O error {}", ex.toString()));

As shown above, creating files in /dev is not allowed. So we expect to see three IOExceptions?

But in actual execution, only the first onErrorContinue is executed, while the second onErrorContinue is ignored, and the call chain terminates early:

Exception in thread "main" reactor.core.Exceptions$ErrorCallbackNotImplemented: java.io.IOException: Operation not permitted
Caused by: java.io.IOException: Operation not permitted
	at java.base/java.io.UnixFileSystem.createFileExclusively(Native Method)
	at java.base/java.io.File.createNewFile(File.java:1026)

If you think this is normal, consider similar code logic but without handling FileNotFoundException. This shouldn't matter since createNewFile() throws a generic IOException. But what's the result?

WARN - I/O error java.io.IOException: Operation not permitted
WARN - I/O error java.io.IOException: Operation not permitted
WARN - I/O error java.io.IOException: Operation not permitted

Honestly, I don't quite understand what's happening here. Why does removing the seemingly ignored FileNotFoundException handling suddenly change the behavior of IOException handling?

Using onErrorResume and doOnError​

I'd rather use the slightly less efficient but more readable onErrorResume() because it's predictable. Notice how I split doOnError() and onErrorResume():

doOnError for logging and metrics, onErrorResume for actual handling of specific exceptions 
public static void main(String[] args) {
  Flux
      .just("one.txt", "two.txt", "three.txt")
      .flatMap(file -> createFile(file))
      .doOnNext(e -> log.info("Got file {}", e))
      .subscribe();
}

private static Mono<Boolean> createFile(String file) {
  return Mono
      .fromCallable(() -> new File("/dev", file).createNewFile())
      .doOnError(FileNotFoundException.class, ex -> log.warn("Not found? {}", ex.toString()))
      .doOnError(IOException.class, ex -> log.warn("I/O error {}", ex.toString()))
      .onErrorResume(IOException.class, ex -> Mono.empty());
}

We get the expected response without the unexpected behavior of onErrorContinue():

WARN I/O error for one.txt: java.io.IOException: Operation not permitted
WARN I/O error for two.txt: java.io.IOException: Operation not permitted
WARN I/O error for three.txt: java.io.IOException: Operation not permitted

So, long story short, onErrorContinue() was created to improve error handling performance.

It achieves this by avoiding wrapping operations in Mono.fromCallable(). Regardless, its behavior is sometimes hard to understand. Additionally, not every upstream operator supports resuming.

If you don't quite understand the sentence above, I would stay away from onErrorContinue(). Actually, I recommend avoiding onErrorContinue() in general.

You can achieve the same effect using onErrorResume() or onErrorReturn().