class ReactorMergeOperations {

    public static <K> Mono<Map<K, Optional<Throwable>>> merge(Map<K, Mono<Void>> monos, int concurrencyLimit, Scheduler scheduler) {
        List<Flux<Pair<K, Optional<Throwable>>>> m2 = new ArrayList<>();
        monos.forEach((key, mono) -> {

            Flux<Pair<K, Optional<Throwable>>> x = mono.toProcessor().ignoreElement().materialize().map(result -> {
                        Optional<Throwable> error = result.getType() == SignalType.ON_ERROR
                                ? Optional.of(result.getThrowable())
                                : Optional.empty();
                        return Pair.of(key, error);
                    }
            ).flux();
            m2.add(x);
        });

        return Flux.merge(Flux.fromIterable(m2), concurrencyLimit)
                .subscribeOn(scheduler)
                .collectList()
                .map(list -> list.stream().collect(Collectors.toMap(Pair::getLeft, Pair::getRight)));
    }

}