Support with loom seems to be broken for kafka binder

[wyhasany]

Kafka running on virtual threads block graceful shutdown.

To Reproduce

1. Add following configuration to one of your example projects:

    @Bean
    @ConditionalOnThreading(Threading.VIRTUAL)
    @SuppressWarnings("rawtypes")
    ListenerContainerCustomizer<?> visibilityListenerContainerCustomizer() {
        return (container, destinationName, _) -> {
            if (container instanceof ConcurrentMessageListenerContainer listenerContainer) {
                SimpleAsyncTaskExecutor executor = new SimpleAsyncTaskExecutor(destinationName + "-");
                executor.setVirtualThreads(true);
                listenerContainer.getContainerProperties().setListenerTaskExecutor(executor);
            }
        };
    }

2. Run app
3. Try to SIGTERM it
4. It won't stop

Version of the framework
Spring Boot 3.2.2, Spring Cloud 2023.0.0, Spring Cloud Stream 4.1.0, Java 21
Expected behavior
Application should stop timely as it happens for platform threads.

Screenshots

Additional context
I think that documentation should be improved for using virtual threads if it is supported or not. I found confusing
Spring Boot migration documentation, I supposed you will use auto configured factories from there. If it is not used
you should recommend to users to disable them:

spring.autoconfigure.exclude=org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration,\
  org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration

I tested also Rabbit and it works seamlessly with following configuration:

    @Bean
    @ConditionalOnThreading(Threading.VIRTUAL)
    ListenerContainerCustomizer<MessageListenerContainer> listenerContainerCustomizer() {
        return (container, destinationName, _) -> {
            if (container instanceof AbstractMessageListenerContainer listenerContainer) {
                var taskExecutor = new SimpleAsyncTaskExecutor(destinationName + "-");
                taskExecutor.setVirtualThreads(true);
                listenerContainer.setTaskExecutor(taskExecutor);
            }
        };
    }

[artembilan]

See more info here: spring-projects/spring-kafka#3013.
Yes, for now it is better to not use virtual threads when Kafka client is involved.

[artembilan]

Has nothing to do with Spring Cloud Stream.
See related Spring for Apache Kafka issue: spring-projects/spring-kafka#3013.

But apparently the culprit synchronized is there in the Apache Kafka client by itself.