title: Reactor Flux timeout with RabbitMq MessageListenerContainer
body: <p>Using a hot publisher model, the following timeout exception is occurring roughly 50% of the time in a cloud environment:</p>
  <pre><code>[ERROR] reactor.core.scheduler.Schedulers                  - Scheduler worker in group main failed with an uncaught exception
  reactor.core.Exceptions$ErrorCallbackNotImplemented: java.util.concurrent.TimeoutException: Did not observe any item or terminal signal within 300000ms (and no fallback has been configured)
  Caused by: java.util.concurrent.TimeoutException: Did not observe any item or terminal signal within 300000ms (and no fallback has been configured)
  reactor.core.publisher.FluxTimeout$TimeoutMainSubscriber.handleTimeout(FluxTimeout.java:289)
  reactor.core.publisher.FluxTimeout$TimeoutMainSubscriber.doTimeout(FluxTimeout.java:274)
  reactor.core.publisher.FluxTimeout$TimeoutTimeoutSubscriber.onNext(FluxTimeout.java:396)
  reactor.core.publisher.StrictSubscriber.onNext(StrictSubscriber.java:89)
  datadog.trace.instrumentation.reactor.core.TracingSubscriber.onNext(TracingSubscriber.java:75)
  reactor.core.publisher.FluxOnErrorResume$ResumeSubscriber.onNext(FluxOnErrorResume.java:73)
  datadog.trace.instrumentation.reactor.core.TracingSubscriber.onNext(TracingSubscriber.java:75)
  reactor.core.publisher.MonoDelay$MonoDelayRunnable.run(MonoDelay.java:117)
  reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:68)
  reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:28)
  java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
  java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
  java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
  java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
  java.base/java.lang.Thread.run(Thread.java:834)
  </code></pre>
  <p>However I cannot replicate this locally. My leading theory is the the cause is occurring during the creation of the flux and something sour is happening between <code>Flux</code> and <code>MessageListenerContainer</code>. The creation looks like:</p>
  <pre><code>@Bean
  public Flux&lt;RabbitEventPublishEnvelope&gt; masterFlux(
          Queue eventQueue,
          ObjectMapper objectMapper,
          MessageListenerContainerFactory messageListenerContainerFactory) {
      log.info(&quot;Create a listener for the topic queue: '{}'&quot;, eventQueue.getName());
      MessageListenerContainer mlc = messageListenerContainerFactory
              .createDirectMessageListenerContainer(eventQueue.getName());
  
      log.info(&quot;Define the master Flux for event subscriptions on queue '{}'&quot;, eventQueue.getName());
      Flux&lt;RabbitEventPublishEnvelope&gt; masterFlux = Flux.create(emitter -&gt; {
          mlc.setupMessageListener(m -&gt; {
              RabbitEventPublishEnvelope payload = null;
              try {
                  log.info(&quot;Creating payload&quot;);
                  payload = objectMapper.readValue(m.getBody(), RabbitEventPublishEnvelope.class);
              } catch (IOException e) {
                  log.error(&quot;Failed to parse RabbitEventPublishEnvelope:\n{}&quot;, m.getBody());
                  throw new RuntimeException(&quot;Failed to parse RabbitEventPublishEnvelope&quot;, e);
              } catch (Exception e) {
                  log.error(&quot;Unhandled exception in Flux.create(): {}&quot;, e.getMessage(), e);
              }
              log.info(&quot;Emitting payload&quot;);
              emitter.next(payload);
          });
          emitter.onRequest(v -&gt; {
              log.info(&quot;MLC starting&quot;);
              mlc.start();
              log.info(&quot;Start recipe event subscription&quot;);
          });
          emitter.onDispose(() -&gt; {
              // WARNING: DO NOT issue `mlc.stop();` here or it will cause responses to hang.
              // The main reason this callback handler is implemented is to document what will break our implementation.
              log.info(&quot;Done with recipe event subscription&quot;);
          });
      });
  
      log.info(&quot;Created master flux for queue = '{}'&quot;, eventQueue.getName());
      return masterFlux
              .log(&quot;Publishing flux&quot;)
              .publish()
              .autoConnect()
              .timeout(Duration.ofMillis(10000))
              .doOnError(error -&gt; log.error(&quot;Unhandled exception in masterFlux(): {}&quot;, error.getMessage(), error))
              .log(&quot;Auto connection successful&quot;);
  }
  </code></pre>
  <p>This exception feels like a configuration mistake, but I cannot narrow down what would be causing this timeout to confirm that. Any help would be appreciated!</p>