Reactor Flux timeout with RabbitMq MessageListenerContainer

Using a hot publisher model, the following timeout exception is occurring roughly 50% of the time in a cloud environment:

[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)

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 Flux and MessageListenerContainer. The creation looks like:

@Bean
public Flux<RabbitEventPublishEnvelope> masterFlux(
        Queue eventQueue,
        ObjectMapper objectMapper,
        MessageListenerContainerFactory messageListenerContainerFactory) {
    log.info("Create a listener for the topic queue: '{}'", eventQueue.getName());
    MessageListenerContainer mlc = messageListenerContainerFactory
            .createDirectMessageListenerContainer(eventQueue.getName());

    log.info("Define the master Flux for event subscriptions on queue '{}'", eventQueue.getName());
    Flux<RabbitEventPublishEnvelope> masterFlux = Flux.create(emitter -> {
        mlc.setupMessageListener(m -> {
            RabbitEventPublishEnvelope payload = null;
            try {
                log.info("Creating payload");
                payload = objectMapper.readValue(m.getBody(), RabbitEventPublishEnvelope.class);
            } catch (IOException e) {
                log.error("Failed to parse RabbitEventPublishEnvelope:\n{}", m.getBody());
                throw new RuntimeException("Failed to parse RabbitEventPublishEnvelope", e);
            } catch (Exception e) {
                log.error("Unhandled exception in Flux.create(): {}", e.getMessage(), e);
            }
            log.info("Emitting payload");
            emitter.next(payload);
        });
        emitter.onRequest(v -> {
            log.info("MLC starting");
            mlc.start();
            log.info("Start recipe event subscription");
        });
        emitter.onDispose(() -> {
            // 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("Done with recipe event subscription");
        });
    });

    log.info("Created master flux for queue = '{}'", eventQueue.getName());
    return masterFlux
            .log("Publishing flux")
            .publish()
            .autoConnect()
            .timeout(Duration.ofMillis(10000))
            .doOnError(error -> log.error("Unhandled exception in masterFlux(): {}", error.getMessage(), error))
            .log("Auto connection successful");
}

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!