+ public Flux<String> restartEiJobs(EiProducer producer, EiJobs eiJobs) {
+ final int maxNoOfParalellRequests = 10;
+ Retry retrySpec = Retry.backoff(3, Duration.ofSeconds(1));
+
+ return Flux.fromIterable(producer.getEiTypes()) //
+ .flatMap(type -> Flux.fromIterable(eiJobs.getJobsForType(type))) //
+ .flatMap(job -> postStartEiJob(producer, job, retrySpec), maxNoOfParalellRequests) //
+ .onErrorResume(t -> {
+ logger.error("Could not restart EI Job for producer: {}, reason :{}", producer.getId(), t.getMessage());
+ return Flux.empty();
+ }); //
+
+ }
+
+ private Mono<String> postStartEiJob(EiProducer producer, EiJob eiJob, Retry retrySpec) {