Why does my Flux stream running in parallel skipping some records?

Question

I am trying to understand Flux.parallel() working. My requirement is as follows: There is a list of String stringList, and let's say we are doing some task on each string in parallel and one of the string takes some time, for instance here I am using Thread.sleep, then I want to collect the string in Flux<String>, but before this I need to make sure if the execution for all the strings in parallel is finished.

In the following code, it is skipping the string Mango, why is that? And if I use blockLast() then it is not skipping but if I use blockLast(), it won't give return Flux<String>. Also, What is the difference between sequential and blockLast?

List<String> list = Arrays.asList("Mango", "Apple", "Grapes", "Java");
Flux<String> flust = Flux.fromIterable(list)
  .parallel(10)
  .runOn(Schedulers.parllel(), list.size())
  .map(string -> {
     if(string.equals("Mango")) {
        try {
          Thread.sleep(2000);
        } catch(InterruptedException e) {
          e.printStackTrace();
        }
      }
      //some task
      System.out.println(Thread.currentThread().getName() + " " + string);
      return string;
    })
   .sequential();
flust.subscribe(System.out::println);

OUTPUT :

parallel-2 Apple

parallel-4 Java

parallel-3 Grapes

Apple

Grapes

Java

Does this answer your question? [Custom thread pool in Java 8 parallel stream](https://stackoverflow.com/questions/21163108/custom-thread-pool-in-java-8-parallel-stream) — K.Nicholas, Jan 24 '22 at 15:10
@K.Nicholas No it doesn't the example is of Java Stream, here I am using Reactive Stream. Also, both of the streams have different operators. — Java Programmer, Jan 24 '22 at 15:13
could you please explain more about `Schedulers.fromExecutorService(process.getExecutor())` what is `process` and what does `getExecutor` return — Toerktumlare, Jan 24 '22 at 16:53
Is the current thread blocking until the subscriber has completed processing? When I run with .parallel().runOn(Schedulers.parallel()) with a Thread.sleep(2000) at the end I see all fruit. — Lee Greiner, Jan 24 '22 at 19:14
@Toerktumlare process is nothing but a component, and process.getExecutor() just return a ThreadPoolExecutor manually configured. — Java Programmer, Jan 25 '22 at 05:35
Well show us the missing code as this code can not be run and i cant reproduce your claims using the standard parallell executor. — Toerktumlare, Jan 25 '22 at 07:21
@Toerktumlare I have edited the code, Still the result it same. — Java Programmer, Jan 25 '22 at 07:30

lkatiforis · Accepted Answer · 2022-01-26T08:15:29.430

As mentioned in the Reactor javadoc subscribe method:

Keep in mind that since the sequence can be asynchronous, this will immediately return control to the calling thread. This can give the impression the consumer is not invoked when executing in a main thread or a unit test for instance.

This means that the end of the main method is reached, and thus the main thread exits before the Mango element is consumed.

To solve this, you can use blockLast() instead of subscribe() and trigger the side-effect (printing) within doOnNext method:

    Flux<String> flust = Flux.fromIterable(list)
        .parallel(10)
        .runOn(Schedulers.parallel(), list.size())
        .map(string -> {
            if (string.equals("Mango")) {
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            //some task
            System.out.println(Thread.currentThread().getName() + " " + string);
            return string;
        })
        .sequential()
        .doOnNext(System.out::println);

    flust.blockLast();

Here, you block execution on the main thread until the last element on the Flux is processed.

Yes, blockLast() will do the work, But I want to understand why is this. happening in the first place, why is it skipping "Mango"? — Java Programmer, Jan 25 '22 at 08:25

Why does my Flux stream running in parallel skipping some records?

1 Answers1