Concurrency and Virtual Threads

[jhg023]

My service creates a ConcurrentKafkaListenerContainerFactory that's used by multiple methods marked with @KafkaListener.

I recently switched my service over to use virtual threads, and enabled them for Kafka as well:

@Bean
public <V> ConcurrentKafkaListenerContainerFactory<String, V> kafkaListenerContainerFactory(
    ObjectMapper objectMapper
) {
    var factory = new ConcurrentKafkaListenerContainerFactory<String, V>();

    factory.setAutoStartup(true);
    factory.setConsumerFactory(kafkaConsumerFactory(objectMapper));

    factory.setContainerCustomizer(container -> {
        container.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        container.getContainerProperties().setListenerTaskExecutor(createVirtualThreadAsyncTaskExecutor());
    });

    return factory;
}

private AsyncTaskExecutor createVirtualThreadAsyncTaskExecutor() {
    var asyncTaskExecutor = new SimpleAsyncTaskExecutor();

    asyncTaskExecutor.setConcurrencyLimit(SimpleAsyncTaskExecutor.UNBOUNDED_CONCURRENCY);
    asyncTaskExecutor.setVirtualThreads(true);

    return asyncTaskExecutor;
}

With this pattern, should I be invoking factory.setConcurrency, or is that not needed anymore?

[garyrussell]

You still need to set concurrency if you want to consume multiple records from different partitions concurrently; each consumer will get its own virtual thread.

[jhg023]

Thanks, Gary!

[Liveitabhi14]

Is it necessary to set the listenerTaskExecutor by ourselves in the ConcurrentKafkaListenerContainer to use virtual threads?
Shouldn't the property spring.threads.virtual.enabled=true do it instead?

[sobychacko]

If you are using Spring Boot, you can use the property spring.thread.virtual.enabled and get a container auto-configured with virtual threads enabled. See this Spring Boot issue for more details: spring-projects/spring-boot#36396. Behind the scenes, Boot does the same in the programmatic way shown above.