Repository navigation
Async @KafkaListener ack mode. Should it be MANUAL? What's its best use? #4500
|
Hi Spring Team, I wanted to ask you if I'm getting this page correctly. It says Trying this small setup locally (4.0.6) it shows as BATCH @Component
public class AsyncListener {
@KafkaListener(topics = "my-topic", groupId = "test-g", id = "test")
public CompletableFuture<String> listen(String message) {
return CompletableFuture.completedFuture(message);
}
}and logging the props @Component
public class KafkaConfig {
private static final Logger log = LoggerFactory.getLogger(KafkaConfig.class);
private final KafkaListenerEndpointRegistry registry;
public KafkaConfig(KafkaListenerEndpointRegistry registry) {
this.registry = registry;
}
@EventListener(ApplicationReadyEvent.class)
public void logContainerProps() {
MessageListenerContainer container = registry.getListenerContainer("test");
var containerProperties = container.getContainerProperties();
log.info("Kafka listener runtime check: containerClass={}", container.getClass().getName());
log.info("Kafka listener runtime check: asyncAcks={}", containerProperties.isAsyncAcks());
log.info("Kafka listener runtime check: ackMode={}", containerProperties.getAckMode());
log.info("Kafka listener runtime check: running={}", container.isRunning());
}
}LOGS I was expecting to see Also in the documentation there not really much describing this behaviour, I can't find a why to this approach, pros vs cons, do non-blocking retries work the same... etc... Thank you, Alex. |
Replies: 1 comment
|
You are checking the configured properties, not the effective acknowledgment mode used by the running In 4.0.6, the listener adapter detects
That happens inside while the running consumer still uses the equivalent of A behavioral test is a better way to verify it: return an incomplete future and check that the offset is not committed until that future completes. If a later record completes first, its commit is deferred until the missing earlier acknowledgment arrives. One important detail in the sample is that public CompletableFuture<Void> listen(String message) {
return CompletableFuture.runAsync(() -> process(message), kafkaWorkerExecutor);
}This model is useful for I/O-bound work when blocking the consumer thread is undesirable, but it has trade-offs:
Those last three points are the documented Non-blocking retry with |
You are checking the configured properties, not the effective acknowledgment mode used by the running
ListenerConsumer.In 4.0.6, the listener adapter detects
CompletableFuturefrom the declared method return type and marks the listener as async:HandlerAdapter. The container then derives a private effective mode:asyncRepliesbecomes true for that adapter.determineAckMode()reads the configured mode and locally changes it toMANUALwhen the listener is async.That happens inside
ListenerConsumerand insidedetermineAckMode(). It does not mutate the publicContainerPropertiesobject. Therefore these values are expected: