spring-kafka delivery attempt is not being increased when message was retried

user_2440613 0
Viewed 86

I am using spring-kafka 2.8.6 with retry RetryTopicConfiguration.

@KafkaListener(
            topics = "...",
            groupId = "...",
            containerFactory = "kafkaListenerContainerFactory")
    public void listenWithHeaders(final @Valid @Payload Event event,
                                  @Header(KafkaHeaders.DELIVERY_ATTEMPT) final int deliveryAttempt) {
}

I have setup common error handler, and also enable delivery attempt header.

@Bean
    public ConcurrentKafkaListenerContainerFactory<String, Event>
    kafkaListenerContainerFactory(@Qualifier("ConsumerFactory") final ConsumerFactory<String, Event> consumerFactory) {
        final ConcurrentKafkaListenerContainerFactory<String, Event> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        factory.setCommonErrorHandler(new DefaultErrorHandler(new ExponentialBackOff(kafkaProperties.getExponentialBackoffInitialInterval(), kafkaProperties.getExponentialBackoffMultiplier())));
        LOGGER.info("setup ConcurrentKafkaListenerContainerFactory");
        factory.getContainerProperties().setDeliveryAttemptHeader(true);
        return factory;
    }

But when retry is triggered, delivery attempt in the message header is always 1, never increase. Do I miss any other part? Thanks!

--- I am using retry topic configuration.

@Bean
    public RetryTopicConfiguration retryableTopicKafkaTemplate(@Qualifier("kafkaTemplate") KafkaTemplate<String, Event> kafkaTemplate) {
        return RetryTopicConfigurationBuilder
                .newInstance()
                .exponentialBackoff(
                        properties.getExponentialBackoffInitialInterval(),
                        properties.getExponentialBackoffMultiplier(),
                        properties.getExponentialBackoffMaxInterval())
                .autoCreateTopics(properties.isRetryTopicAutoCreateTopics(), properties.getRetryTopicAutoCreateNumPartitions(), properties.getRetryTopicAutoCreateReplicationFactor())
                .maxAttempts(properties.getMaxAttempts())
                .notRetryOn(...)    .retryTopicSuffix(properties.getRetryTopicSuffix())
                .dltSuffix(properties.getDltSuffix())
                .create(kafkaTemplate);

---- Followed by Gary's suggestion, have it fully working now with my listener.

 @KafkaListener(
            topics = "...",
            groupId = "...",
            containerFactory = "kafkaListenerContainerFactory")
    public void listenWithHeaders(final @Valid @Payload Event event,
                                  @Header(value = RetryTopicHeaders.DEFAULT_HEADER_ATTEMPTS, required = false) final Integer deliveryAttempt) {
...
spring-kafka
1 Answers

It works fine for me with this:

@SpringBootApplication
public class So72871495Application {

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


    @KafkaListener(id = "so72871495", topics = "so72871495")
    void listen(String in, @Header(KafkaHeaders.DELIVERY_ATTEMPT) int delivery) {
        System.out.println(in + " " + delivery);
        throw new RuntimeException("test");
    }

    @Bean
    public NewTopic topic() {
        return TopicBuilder.name("so72871495").partitions(1).replicas(1).build();
    }

    @Bean
    ApplicationRunner runner(KafkaTemplate<String, String> template,
            AbstractKafkaListenerContainerFactory<?, ?, ?> factory) {

        factory.getContainerProperties().setDeliveryAttemptHeader(true);
        factory.setCommonErrorHandler(new DefaultErrorHandler(new FixedBackOff(5000L, 3)));
        return args -> {
            template.send("so72871495", "foo");
        };
    }

}

foo 1
foo 2
foo 3
foo 4

If you can't figure out what's different for you, please provide an MCRE so I can see what's wrong.

EDIT

With @RetryableTopic, that header is always 1 because each delivery is the first attempt from a different topic.

Use this instead

void listen(String in, @Header(name = RetryTopicHeaders.DEFAULT_HEADER_ATTEMPTS, required = false) Integer attempts) {

Integer not int. It will be null on the first attempt and 2, 3, etc on the retries.

user_12407630
Related