Reactor-core: ArrayIndexOutOfBoundsException on flattening Flux<Mono.publish(f)>

[hossameldeen]

Simplest test case that throws the exception

Mono<Integer> simpleMono = Mono.just(1);
Mono<Integer> publishMulticastMono = simpleMono.publish(m -> m);
Flux<Mono<Integer>> fluxThatHasPublishMulticastMonos = Flux.just(publishMulticastMono);

fluxThatHasPublishMulticastMonos
        .flatMap(mono -> mono)  // flatten
        .blockLast();           // fire. Note: No problem with `blockFirst()`

stacktrace

java.lang.ArrayIndexOutOfBoundsException: 0

at reactor.core.publisher.FluxPublishMulticast$FluxPublishMulticaster.drainSync(FluxPublishMulticast.java:308)
at reactor.core.publisher.FluxPublishMulticast$FluxPublishMulticaster.drain(FluxPublishMulticast.java:250)
at reactor.core.publisher.FluxPublishMulticast$FluxPublishMulticaster.onSubscribe(FluxPublishMulticast.java:181)
at reactor.core.publisher.MonoJust.subscribe(MonoJust.java:54)
at reactor.core.publisher.FluxPublishMulticast.subscribe(FluxPublishMulticast.java:90)
at reactor.core.publisher.MonoSource.subscribe(MonoSource.java:65)
at reactor.core.publisher.FluxFlatMap.trySubscribeScalarMap(FluxFlatMap.java:192)
at reactor.core.publisher.FluxFlatMap.subscribe(FluxFlatMap.java:110)
at reactor.core.publisher.Flux.blockLast(Flux.java:1716)
at com.orange.odu.prepaid.proxy.unitTest.ReactorTest.testBranchingOfMonoInsideFluxSimplified(ReactorTest.java:150)

Similar test cases that pass

Mono<Integer> simpleMono = Mono.just(1);
Mono<Integer> publishMulticastMono = simpleMono.publish(m -> m);
publishMulticastMono.block();

Mono<Integer> simpleMono = Mono.just(1);
Mono<Integer> publishMulticastMono = simpleMono.publish(m -> m);
Flux<Mono<Integer>> fluxThatHasPublishMulticastMonos = Flux.just(publishMulticastMono);

fluxThatHasPublishMulticastMonos
        .flatMap(mono -> mono)  // flatten
        .blockFirst();           // fire. This time, `blockFirst()` instead of `blockLast()`

Versions tested

3.0.3.RELEASE
3.0.4.RELEASE
3.0.5.RELEASE

Code tracing

This part is probably just noise and not even that hard to know, I'm just trying to be helpful. Feel free to ignore it.

In FluxPublishMulticast::drainSync():

• a[i].actual is a MonoNext, s of which is a PublishClientSubscription whose parent is the original FluxPublishMulticast.
• in while(e != r):
  
  
  
  • In the first iteration, a[i].actual.onNext(v); calls s.cancel(), so removeAndDrain is called on the original FluxPublishMulticast, which sets subscribes to an empty array.
  
  
  • In the second iteration, n would remain one, while subscribers became an empty array.
  
  

i.e., completion is trying to be done twice, I think. Once, because of MonoNext::onNext() knows it has only one element to emit, and another in the next iteration of FluxPublishMulticast when it finds no more elements in its queue. Maybe, the first iteration of the while(e != r) should've cancelled the FluxPublishMulticast in this case.
That's what I understood from the code, that is. May be totally wrong.

I really wanted to help more but can't understand what wip, missed & FluxFlatMap::trySubscribeScalarMap() are. Was hoping for this to be my 1st meaningful PR but seems like the dream is gone :cry:. I now hope anything I've said is correct :joy:. Remember: You've been warned to not read this part :P.

[akarnokd]

The problem is here, not re-reading the array length. I can't remember why I didn't use for-each loop back then.

[smaldini]

We had a similar issue in publish but i didn't double check the multicast one my bad !
👍 Awesome report btw @hossameldeen !

[smaldini]

A 3.0.6.BUILD-SNAPSHOT should be out in a couple minutes with the fix included, please reopen @hossameldeen if you still observe the weird behavior.

[abdelrahmanmohamed]

@smaldini
me and @hossameldeen still waiting for the snapshot to updated