Reactor - Group By, Parallel execution of different groups

I am trying to use Spring Project Reactor to achieve the following behavior:

1. Based on an List, I want to do grouping based on a property
2. I want to make sure that the processing happens sequentially WITHIN a group, but parallel BETWEEN groups.

How can this be achieved? I have seen several similar questions on SO, but none of them answer my question and the examples are not working properly.

List<CurrencyWithValue> pairWithValueList = Arrays.asList(
                new CurrencyWithValue("EUR", 1.0),
                new CurrencyWithValue("USD", 1.0),
                new CurrencyWithValue("CHF", 1.0),
                new CurrencyWithValue("SEK", 1.0),
                new CurrencyWithValue("SEK", 2.0),
                new CurrencyWithValue("EUR", 2.0),
                new CurrencyWithValue("USD", 2.0),
                new CurrencyWithValue("CHF", 2.0),
                new CurrencyWithValue("SEK", 3.0),
                new CurrencyWithValue("EUR", 3.0),
                new CurrencyWithValue("USD", 3.0),
                new CurrencyWithValue("CHF", 3.0),
                new CurrencyWithValue("EUR", 4.0),
                new CurrencyWithValue("USD", 4.0),
                new CurrencyWithValue("CHF", 4.0),
                new CurrencyWithValue("SEK", 4.0),
                new CurrencyWithValue("SEK", 5.0)
        );


Flux.fromIterable(pairWithValueList).groupBy(v -> v.getCurrency())
.flatMap(f -> f)
.parallel(3)
.runOn(Schedulers.newBoundedElastic(3, 1000, "k-task"))
.subscribe(i -> System.out.println("Thread:" + Thread.currentThread().getName() + "::Data:" + i.getCurrency() + "::" + i.getAmount()));

The results are kinda different on every execution - sometimes the println happens in order, sometimes not.
Here is the result of one run:

Thread:k-task-3::Data:CHF::1.0
Thread:k-task-2::Data:USD::1.0
Thread:k-task-1::Data:EUR::1.0
Thread:k-task-2::Data:SEK::2.0
Thread:k-task-3::Data:EUR::2.0
Thread:k-task-2::Data:CHF::2.0
Thread:k-task-1::Data:SEK::1.0
Thread:k-task-2::Data:USD::3.0
Thread:k-task-3::Data:SEK::3.0
Thread:k-task-2::Data:USD::4.0
Thread:k-task-1::Data:USD::2.0
Thread:k-task-2::Data:SEK::5.0
Thread:k-task-3::Data:CHF::3.0
Thread:k-task-1::Data:EUR::3.0
Thread:k-task-3::Data:CHF::4.0
Thread:k-task-1::Data:EUR::4.0
Thread:k-task-1::Data:SEK::4.0

I have tried replacing flatMap with concatMap or flatMapSequential but the result is still the same.
I am new to Project Reactor and I am struggling to achieve my desired behavior.

Any ideas how i can solve this?
Thanks a lot

First of all, you need to learn how to use the project reactor properly.

For example, the groups created by groupBy are immediately flattened, so technically it is similar to adding an effect by an operator, and then the next operator removing it right away so the previous was not there at all.

Then, parallel is working properly but it does not offer you any guarantee that all the same currencies are processed sequentially.

To solve your problem you just need the groupBy operator and publishOn, instead of parallel + runOn:

Flux.fromIterable(pairWithValueList)
    .groupBy(v -> v.getCurrency())
    .flatMap(f -> 
         f.publishOn(Schedulers.parallel())
          .doOnNext(i -> System.out.println("Thread:" + Thread.currentThread().getName() + "::Data:" + i.getCurrency() + "::" + i.getAmount())
    )
    .subscribe();

The above solution is going to work, however, there are multiple caveats here.

The problem that may emerge is the concurrency limit that is present in the flatMap operator. By default, concurrency is set to 256, so you need to make sure that the number of allocated groups by groupBy is less or equal to that limit. Otherwise, you have to adjust the concurrency of flatMap and set it equal to the number of possible currencies.

If the number of currencies is unpredictable (let's assume it is an infinite number) then you need to define a different grouping mechanism. For example, you can use the first letter instead to create a group, (or calculate some hash code, and then do as follows hash % 256):

Flux.fromIterable(pairWithValueList)
    .groupBy(v -> v.getCurrency().charAt(0))
    .flatMap(f -> 
         f.publishOn(Schedulers.parallel())
          .doOnNext(i -> System.out.println("Thread:" + Thread.currentThread().getName() + "::Data:" + i.getCurrency() + "::" + i.getAmount())
    )
    .subscribe();

Other caveats

1. When you do grouping as above, you have to make sure that all the groups have more or less the same number of elements within them over time. Make sure your grouping algorithm is balanced
2. Schedulers.parallel() has a limited number of threads equal to a number of cores available. You may want to create your own dedicated scheduler or use boundedElastic instead