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!