Auto Config Non-Blocking Retries with asyncAcks

[wapkch]

In what version(s) of Spring for Apache Kafka are you seeing this issue?

3.0.7

Describe the bug

when using asyncAcks together with RetryableTopic, the retried message will not be invoked on KafkaListener.

To Reproduce

@RestController
@SpringBootApplication
public class SpringkafkaAsyncacksDemoApplication {

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

    @Autowired
    private KafkaTemplate<Object, Object> template;

    @PostMapping(path = "/send/{what}")
    public void sendFoo(@PathVariable String what) {
        this.template.send("topic1", what);
    }

    @KafkaListener(id = "fooGroup", topics = "topic1")
    @RetryableTopic(attempts = "5", backoff = @Backoff(delay = 2_000, maxDelay = 10_000, multiplier = 2))
    public void onMessage(ConsumerRecord<String, ?> record, Acknowledgment acknowledgment, Consumer<?, ?> consumer) {
        if (record.value().equals("fail")) {
            throw new RuntimeException("failed");
        }
        acknowledgment.acknowledge();
    }

}

application.properties

spring.kafka.listener.async-acks=true
spring.kafka.retry.topic.enabled=true
spring.kafka.listener.ack-mode=manual

1. Run local kafka server
2. Start the application
3. Access http://localhost:8080/send/fail, and the retried message will not be invoked on KafkaListener.

Expected behavior

I expect Non-Blocking Retries can work with asyncAcks.

I konw using asyncAcks in above case is meaningless, but sometimes the async processing in @KafkaListener is useful. For example:

@Component
public class FooConsumer {

    @KafkaListener(id = "fooGroup", topics = "topic1")
    @RetryableTopic(attempts = "5", backoff = @Backoff(delay = 2_000, maxDelay = 10_000, multiplier = 2))
    public void onMessage(ConsumerRecord<String, ?> record, Acknowledgment acknowledgment, Consumer<?, ?> consumer) {
       threadPoolTaskExecutor.submit(() -> {
           try {
                 doConsume(record);
                 acknowledgment.acknowledge();
           } catch (Exception ex) {
                 container.getCommonErrorHandler().handleOne(new RuntimeException(), record, consumer, container);
           }
        });
    }

}

Sample

https://github.com/wapkch/springkafka-asyncacks-demo

[garyrussell]

I will look at your example but, with this code:

       threadPoolTaskExecutor.submit(() -> {
           try {
                 doConsume(record);
                 acknowledgment.acknowledge();
           } catch (Exception ex) {
                 container.getCommonErrorHandler().handleOne(new RuntimeException(), record, consumer, container);
           }
        });

You must call acknowledgment.acknowledge(); after the error handler handles the error.

[garyrussell]

@wapkch After further review, the use of asyncAcks with @RetryableTopic requires the error handler seekAfterError property to be false. We need to document this limitation.

In addition, the built-in DLT handler is not compatible with MANUAL acks.

See the following changes I made to your example app to make it work...

@RestController
@SpringBootApplication
public class SpringkafkaAsyncacksDemoApplication {

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

    @Autowired
    private KafkaTemplate<Object, Object> template;

    @GetMapping(path = "/send/{what}")
    public void sendFoo(@PathVariable String what) {
        this.template.send("topic1", what);
    }
    
}

@Component
class Listener {

    @KafkaListener(id = "fooGroup", topics = "topic1")
    @RetryableTopic(attempts = "2", backoff = @Backoff(delay = 2_000, maxDelay = 2_000, multiplier = 2))
    public void onMessage(ConsumerRecord<String, ?> record, Acknowledgment acknowledgment, Consumer<?, ?> consumer) {
        System.out.println(record.topic() + ":" + record.value());
        if (record.value().equals("fail")) {
            throw new RuntimeException("failed");
        }
        acknowledgment.acknowledge();
    }

    @DltHandler
    void dlt(ConsumerRecord<String, ?> record, Acknowledgment acknowledgment, Consumer<?, ?> consumer) {
        System.out.println(record.topic() + ":" + record.value());
        acknowledgment.acknowledge();
    }

}

@Configuration
class Config extends RetryTopicConfigurationSupport {

    @Override
    protected void configureCustomizers(CustomizersConfigurer customizersConfigurer) {
        customizersConfigurer.customizeErrorHandler(eh -> eh.setSeekAfterError(false));
    }

    @Bean
    TaskScheduler sched() {
        return new ThreadPoolTaskScheduler();
    }

}