Distinct concurrency for retry and DLT topic containers

[jgslima]

I am struggling to understand how to achieve a configuration where I can have the listener container for the retry and DLT topics with a concurrency different from the container of the main topic (the retry and DLT topics do not need the same amount of threads/consumers).

I have two beans of ConcurrentKafkaListenerContainerFactory<?, ?>:

• one with the name kafkaListenerContainerFactory, to be used as the main factory for the main containers.
• another one, dedicated to the retry/DLT containers, configured with the desired concurrency for them.

Then, I already tried to force the use of the second factory for the retry/DTL containers with these 2 ways:

• specifying the factory name in the listenerContainerFactory attribute of the @RetryableTopic annotation.
• forcing the name of this factory to be RetryTopicBeanNames.DEFAULT_LISTENER_CONTAINER_FACTORY_BEAN_NAME.

But in the end, I always end up having, for the retry and DLT topic, an amount of containers/threads equal to the concurrency of the main listener (that is, the concurrency declared in the @KafkaListener annotation).

I tried to understand and debug ListenerContainerFactoryResolver, at some point it seems to resolve to the factory with the customized concurrency, but in the end for some reason the amount of containers/threads are egual to the main listener.

[garyrussell]

Here is a work around, until we can address this:

@SpringBootApplication
public class Kgh2434Application {

	public static void main(String[] args) {
		SpringApplication.run(Kgh2434Application.class, args);
	}

	@KafkaListener(id = "kgh2434", topics = "kgh2434", concurrency = "1", autoStartup = "false")
	@RetryableTopic
	void listen(String in) {
		System.out.println(in);
	}

	@Bean
	ApplicationRunner runner(KafkaListenerEndpointRegistry registry) {
		return args -> {
			((ConcurrentMessageListenerContainer<?, ?>) registry.getListenerContainer("kgh2434")).setConcurrency(6);
			registry.start();
		};
	}

}

[garyrussell]

@jgslima I created a issue for this #2443

Do you think a single concurrency property will suffice that is applied to all the retry containers (including DLT), or do you think a separate concurrency property is needed for each stage of the retries?

It should be obvious that a single override for all containers will be much easier to implement.

[jgslima]

@garyrussell I am very sorry, did not notice your question. Not sure if you have decided yet, but yes, a single concurrency config shoud suffice.
Thanks a lot.

[garyrussell]

No problem; that's what I went with, to avoid adding complexity.

4fedd75

[nkalra0123]

@garyrussell with spring boot version 2.5.12, spring kafka 2.7.12 version, this code is creating an issue.

As we have set factory.setContainerCustomizer(customizer), when we have RetryableTopic and KafkaListener annotations, then configure function from ListenerContainerFactoryConfigurer is not getting called.

public ConcurrentKafkaListenerContainerFactory<?, ?> configure(
			ConcurrentKafkaListenerContainerFactory<?, ?> containerFactory, Configuration configuration) {
		return isCached(containerFactory)
				? containerFactory
				: addToCache(doConfigure(containerFactory, configuration.backOffValues));
	}

because of this backoff from RetryableTopic is not working properly, retries are getting executed immediately, as RecordMessagingMessageListenerAdapter objects are getting created instead of KafkaBackoffAwareMessageListenerAdapter.

But this code works fine with spring boot 2.7.8, with spring kafka 2.8.11 version.

And even with spring boot version 2.5.12 and with spring kafka 2.8.11 version, code is running without any issues.

But acc to the compatibility matrix
https://spring.io/projects/spring-kafka
2.8.x version of spring kafka should be used with 2.6.x or 2.7.x spring boot.

I cannot upgrade the spring boot version. Is it safe to use spring kafka 2.8.11 version with spring boot version 2.5.12?

Can you suggest any other way to change the Concurrency of retry and dlt handlers.

[garyrussell]

I don't know; have you tried it?

Those versions (Boot 2.5.x and SK 2.7.x, 2.8.x) are no longer supported.