After the fix for #3057, DirectMessageListenerContainer cannot be restarted after being stopped via SmartLifecycle.stop(Runnable). The cleanUpTaskScheduler() method nulls the internally-created TaskScheduler, but because stop(Runnable) does not reset the initialized flag, the scheduler is never re-created on the next start().

This manifests as IllegalStateException: taskScheduler must be provided at DirectMessageListenerContainer.startMonitor().

Affected path:

1. Container starts -- doInitialize() creates a default ThreadPoolTaskScheduler, initialized set to true
2. SmartLifecycle.stop(Runnable) is called -- cleanUpTaskScheduler() shuts down and nulls the scheduler. stop(Runnable) calls shutdown(callback) which (by design) does not reset initialized
3. start() is called again -- initialized is true, so afterPropertiesSet() is skipped. doStart() -> actualStart() detects taskScheduler == null and calls afterPropertiesSet(), but initialize() short-circuits on if (!this.initialized)
4. startMonitor() hits Assert.state(this.taskScheduler != null, ...) and throws

The no-arg shutdown() does reset initialized = false, but the shutdown(Runnable) overload does not -- this is intentional for the restartable SmartLifecycle contract. The #3057 fix added cleanUpTaskScheduler() to the stop(Runnable) path, which conflicts with this contract.

This doesn't happen when using the default SimpleMessageListenerContainer.

Regression from 3.2.x: In Spring AMQP 3.2.x, the scheduler cleanup only exists in doStop(), which goes through the stop() -> doStop() -> shutdown() (no-arg) path that does reset initialized. The #3057 fix in 4.0.x added cleanUpTaskScheduler() to the stop(Runnable) path as well, introducing this inconsistency. Spring Boot 3.5.x (Spring AMQP 3.2.x) is not affected.

Reproduction:

This occurs when running a Spring Boot integration test suite where:

• Multiple test classes share a cached application context using DirectMessageListenerContainer (spring.rabbitmq.listener.type=direct)
• Some test classes start a RabbitMQ TestContainer and some do not
• The intermediate tests (without RabbitMQ) trigger SmartLifecycle.stop(Runnable) on the DMLC
• Later tests that restart the container fail with the IllegalStateException

@SpringBootTest
@Testcontainers
class TestA {
    @Container
    @ServiceConnection
    static RabbitMQContainer rabbit = new RabbitMQContainer("rabbitmq:3.13");

    @Test
    void testWithRabbit() {
        // passes -- DMLC starts, scheduler created, initialized = true
    }
}

// Intermediate test class without RabbitMQ -- triggers SmartLifecycle stop(Runnable)
// on the cached context's DMLC. Scheduler is nulled, initialized stays true.

@SpringBootTest
@Testcontainers
class TestC {
    @Container
    @ServiceConnection
    static RabbitMQContainer rabbit = new RabbitMQContainer("rabbitmq:3.13");

    @Test
    void testWithRabbitAgain() {
        // FAILS: IllegalStateException: taskScheduler must be provided
    }
}

Suggested fix:

The self-healing in actualStart() (line 462) calls afterPropertiesSet() which is guarded by initialized. It should instead recreate the scheduler directly:

if (this.taskScheduler == null) {
    ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
    threadPoolTaskScheduler.setThreadNamePrefix(getListenerId() + "-consumerMonitor-");
    threadPoolTaskScheduler.afterPropertiesSet();
    this.taskScheduler = threadPoolTaskScheduler;
}

Workaround:

Provide an external TaskScheduler bean via setTaskScheduler(). When taskSchedulerSet is true, cleanUpTaskScheduler() skips the cleanup entirely, avoiding the issue.

Spring AMQP version: 4.0.0 -- 4.0.2 (regression introduced by #3057). Spring AMQP 3.2.x is not affected.