Reproduce reactor 3 thread stealing when flatMapIterable is followed by publishOn

ReactorThreadStealingTest.java

import java.time.Duration;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.LockSupport;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import org.junit.jupiter.api.Test;

import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;
import reactor.core.scheduler.Schedulers;

class ReactorThreadStealingTest {

  @Test
  public void reproducer() throws Throwable {
    AtomicReference<Throwable> throwable = new AtomicReference<>();
    CountDownLatch latch = new CountDownLatch(1);

    Flux.<Integer>create(emitter -> infiniteFastProducer(emitter), FluxSink.OverflowStrategy.BUFFER)
        .take(Duration.ofSeconds(5))
        .publishOn(Schedulers.newParallel("process"))
        .flatMapIterable(this::process)
        .publishOn(Schedulers.newParallel("downstream"))
        .subscribe(
            x -> LockSupport.parkNanos(1_000),
            e -> {
              throwable.set(e);
              latch.countDown();
            },
            () -> latch.countDown());

    latch.await();
    if (throwable.get() != null)
      throw throwable.get();
  }

  private List<Integer> process(Integer x) {
    LockSupport.parkNanos(100_000);
    if (Thread.currentThread().getName().contains("downstream")) {
      final RuntimeException e = new RuntimeException("This should not run on a 'downstream' thread");
      e.printStackTrace();
      throw e;
    }
    return IntStream.range(0, 100).map(i -> i*x).boxed().collect(Collectors.toList());
  }

  private void infiniteFastProducer(FluxSink<Integer> emitter) {
    final AtomicBoolean stopped = new AtomicBoolean(false);
    new Thread(() -> {
      int i = 0;
      while (!stopped.get()) {
        LockSupport.parkNanos(1_000);
        if (emitter.requestedFromDownstream() > 0)
          emitter.next(++i);
      }
    }).start();
    emitter.onCancel(() -> stopped.set(true));
    emitter.onDispose(() -> stopped.set(true));
  }

}

Test case with io.projectreactor:reactor-core:3.3.2.REALEASE fails:

java.lang.RuntimeException: This should not run on a 'downstream' thread
	at ReactorThreadStealingTest.process(ReactorThreadStealingTest.java:44)
	at reactor.core.publisher.FluxFlattenIterable$FlattenIterableSubscriber.drainAsync(FluxFlattenIterable.java:341)
	at reactor.core.publisher.FluxFlattenIterable$FlattenIterableSubscriber.drain(FluxFlattenIterable.java:649)
	at reactor.core.publisher.FluxFlattenIterable$FlattenIterableSubscriber.request(FluxFlattenIterable.java:273)
	at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.runAsync(FluxPublishOn.java:405)
	at reactor.core.publisher.FluxPublishOn$PublishOnSubscriber.run(FluxPublishOn.java:484)
	at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:84)
	at reactor.core.scheduler.WorkerTask.call(WorkerTask.java:37)
	at java.util.concurrent.FutureTask.run(FutureTask.java:266)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)

Same result with 3.3.6.RELEASE