RetryTopicConfigurationBuilder#includeTopics does not work with the List

[hariraogotit]

@garyrussell

I tried the following for non-blocking retry

My listener

@KafkaListener(topics = {"${redacted-request-topic}"})
  public void redacted(ConsumerRecord<String, byte[]> record,
                                       @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                                       @Header(KafkaHeaders.OFFSET) long offset) throws Exception {

@Bean
 public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, byte[]> kafkaTemplate) {
   return RetryTopicConfigurationBuilder
           .newInstance()
           .includeTopics(List.of("request-topic"))
           .exponentialBackoff(3000, 2, 100000)
           .maxAttempts(5)
           .retryTopicSuffix("-asl-nm.retry")
           .dltSuffix("-asl-nm.dlt")
           .create(kafkaTemplate);
 }

request-topic is the name of the topic. "${redacted-request-topic}" resolves to request-topic

When the includeTopics is commented out, it is working as expected i.e the retry topic name, dlt name and exponential back off kicks in but when includeTopics is uncommented none of these work and it retries in blocking mode.

I can include the logs but have to redact some information. I will try to share redacted logs if they are really required. Kindly let me know what I have missed here.

I am using spring kafka 2.8.11

I see a similar issue - #1916 and I am sorry I commented in #1920

[garyrussell]

This doesn't work for me with, or without, the includeTopics()...

@SpringBootApplication
public class Kgh2676Application {

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

	@KafkaListener(id = "kgh2676", topics = "${kgh2676}")
	void listen(String in) {
		System.out.println(in);
	}

	@Bean
	RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, byte[]> kafkaTemplate,
										 @Value("${kgh2676}") String topic) {
		return RetryTopicConfigurationBuilder
				.newInstance()
				.includeTopics(List.of(topic))
				.exponentialBackoff(3000, 2, 100000)
				.maxAttempts(5)
				.retryTopicSuffix("-asl-nm.retry")
				.dltSuffix("-asl-nm.dlt")
				.create(kafkaTemplate);
	}

}

The problem being that the myRetryTopic bean is not fully formed yet, while the @KafkaListener is being processed, and therefore can't be discovered.

Moving the RetryTopicConfiguration bean to a different class works for me, even with the includeTopics...

@SpringBootApplication
public class Kgh2676Application {

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

	@KafkaListener(id = "kgh2676", topics = "${kgh2676}")
	void listen(String in) {
		System.out.println(in);
	}

}

@Configuration
class OtherConfiguration
{
	@Bean
	RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, byte[]> kafkaTemplate,
										 @Value("${kgh2676}") String topic) {
		return RetryTopicConfigurationBuilder
				.newInstance()
				.includeTopics(List.of(topic))
				.exponentialBackoff(3000, 2, 100000)
				.maxAttempts(5)
				.retryTopicSuffix("-asl-nm.retry")
				.dltSuffix("-asl-nm.dlt")
				.create(kafkaTemplate);
	}

}

kgh2676=myTopic

kgh2676: partitions assigned: [myTopic-0]
kgh2676-asl-nm.dlt: partitions assigned: [myTopic-asl-nm.dlt-0]
kgh2676-asl-nm.retry-6000: partitions assigned: [myTopic-asl-nm.retry-6000-0]
kgh2676-asl-nm.retry-24000: partitions assigned: [myTopic-asl-nm.retry-24000-0]
kgh2676-asl-nm.retry-3000: partitions assigned: [myTopic-asl-nm.retry-3000-0]
kgh2676-asl-nm.retry-12000: partitions assigned: [myTopic-asl-nm.retry-12000-0]

[garyrussell]

Another solution is to make the bean static so it is created early...

	@Bean
	static RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, byte[]> kafkaTemplate,
										 @Value("${kgh2676}") String topic) {

[hariraogotit]

Let me give more details

MyListener.java

@KafkaListener(topics = {"${redacted-request-topic}"})
  public void redacted(ConsumerRecord<String, byte[]> record,
                                       @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
                                       @Header(KafkaHeaders.OFFSET) long offset) throws Exception {

WorkerKafkaConsumerConfig.java annotated with @configuration in the class level has the following

@Bean
 public RetryTopicConfiguration myRetryTopic(KafkaTemplate<String, byte[]> kafkaTemplate) {
   return RetryTopicConfigurationBuilder
           .newInstance()
           .includeTopics(List.of("request-topic"))
           .exponentialBackoff(3000, 2, 100000)
           .maxAttempts(5)
           .retryTopicSuffix("-asl-nm.retry")
           .dltSuffix("-asl-nm.dlt")
           .create(kafkaTemplate);
 }

Integration test with embedded kafka has the following

@RunWith(SpringRunner.class)
@SpringBootTest(
    properties = {
        "spring.kafka.consumer.auto-offset-reset=earliest",
        "spring.kafka.consumer.group-id=kafka-test",
        "spring.kafka.bootstrap-servers=${spring.embedded.kafka.brokers}",
        "spring.kafka.producer.value-serializer="
            + "org.apache.kafka.common.serialization.ByteArraySerializer"
    },
    classes = {AppEntry.class, WorkerKafkaConsumerConfig.class})
@ActiveProfiles(profiles = {"test", "basicLogging"})
@DirtiesContext

In the integration test I am producing the message. I am first testing this using my integration test. Based on your above answer I believe the issue might be somewhere here - classes = {AppEntry.class, WorkerKafkaConsumerConfig.class})

I tried making myRetryTopic method static as well but no joy. Any leads will be helpful. Thanks for the efforts so far.

[garyrussell]

Please provide an MRCE so I can play with it locally to see what's going on.

[hariraogotit]

@garyrussell It works now. Hence marking the answer. Thanks for your time on this.