Project Reactor TopicProcessor многопоточность

Question

В следующем примере многопоточности выполняется запись в TopicProcessor из двух разных потоков и чтение из TopicProcessor в двух разных потоках. Однако где-то существует состояние гонки, при котором не все события передаются подписчикам, что приводит к зависанию приложения навсегда в processed.await(). Кто-нибудь понимает, почему?

import reactor.core.publisher.Flux;
import reactor.extra.processor.TopicProcessor;
import reactor.extra.processor.WaitStrategy;

import java.time.Duration;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import static java.util.Arrays.asList;

public class ReactorTopicProcessorSample {

  public static class Producer implements Callable<Void> {

    final String name;
    final List<String> data;
    final CountDownLatch producerCount;
    final TopicProcessor<String> topicProcessor;

    public Producer(String name, List<String> data, CountDownLatch submitted, TopicProcessor<String> topicProcessor) {
      this.name = name;
      this.data = data;
      this.producerCount = submitted;
      this.topicProcessor = topicProcessor;
    }

    @Override
    public Void call() throws Exception {
      producerCount.countDown();
      producerCount.await(); // wait until the other producer is submitted to be sure that they run in different threads
      Flux.fromIterable(data)
          .map(s -> "\"" + s + "\"" + " from thread " + Thread.currentThread().getName())
          .delayElements(Duration.ofMillis(10))
          .subscribe(topicProcessor);
      System.out.println("Submitted " + name + " producer in thread " + Thread.currentThread().getName() + ".");
      return null;
    }
  }

  public static void main(String[] args) throws InterruptedException {
    for (int i = 0; i < 100; i++) { // this sample doesn't hang every time. repeat a few times to make it reproducible
      realMain(args);
      System.out.println("\n--- the previous run was successful. running again ---\n");
    }
  }

  public static void realMain(String[] args) throws InterruptedException {

    List<String> numbers = asList("1", "2", "3", "4", "5", "6", "7", "8");
    List<String> characters = asList("a", "b", "c", "d", "e", "f", "g", "h");

    CountDownLatch producerCount = new CountDownLatch(2);
    CountDownLatch subscriberCount = new CountDownLatch(2);
    CountDownLatch processed = new CountDownLatch(
        (int) subscriberCount.getCount() * (numbers.size() + characters.size()));

    ExecutorService exec = Executors.newFixedThreadPool((int) producerCount.getCount());

    TopicProcessor<String> topicProcessor = TopicProcessor.<String>builder()
        .share(true)
        .name("topic-processor")
        .bufferSize(16)
        .waitStrategy(WaitStrategy.liteBlocking())
        .build();

    Flux<String> flux = Flux.from(topicProcessor)
        .doOnSubscribe(s -> subscriberCount.countDown());

    flux.subscribe(out -> {
      System.out.println("Subscriber in thread " + Thread.currentThread().getName() + " received " + out);
      processed.countDown();
    });

    flux.subscribe(out -> {
      System.out.println("Subscriber in thread " + Thread.currentThread().getName() + " received " + out);
      processed.countDown();
    });

    subscriberCount.await();

    exec.submit(new Producer("number", numbers, producerCount, topicProcessor));
    exec.submit(new Producer("character", characters, producerCount, topicProcessor));

    processed.await();
    exec.shutdown();
    topicProcessor.shutdown();
  }
}

Зависимости

<dependency>
  <groupId>io.projectreactor</groupId>
  <artifactId>reactor-core</artifactId>
  <version>3.3.2.RELEASE</version>
</dependency>
<dependency>
  <groupId>io.projectreactor.addons</groupId>
  <artifactId>reactor-extra</artifactId>
  <version>3.3.2.RELEASE</version>
</dependency>

Пример поведения: подписчик получает только символы или получает только цифры, в результате чего программа ждет в processed.await() вечно. Это происходит не каждый раз, иногда это работает, как ожидалось.