Project Reactor conditional execution

If you want to propagate distinct errors for the native vs target language error cases, you'll need to perform async filtering inside a flatMap:

objectFlux.flatMap(o ->    Mono.just(o)        .filterWhen(languageRepository.exists(...)) //native        .switchIfEmpty(Mono.error(new RuntimeException("Native language doesn't exist"))        .filterWhen(languageRepository.exists(...)) //target        .switchIfEmpty(Mono.error(new RuntimeException("Target language doesn't exist"))    )    .flatMap(wordSetRepository::save);

The async filtering inside the flatMap ensures that if the test doesn't pass, the inner sequence is empty. This in turn allows us to detect the case and propagate the adequate error. If both tests pass, the original o is propagated in the main sequence.

The second flatMap takes it from there, only receiving the elements that passed both filters and saving them in DB.

Note that the first element to not pass the filters will interrupt the whole sequence (but it was the same in the blocking code since an exception was thrown).