Understanding Spring Kafka Message Listener Containers and @KafkaListener Configuration

The article introduces Spring Kafka’s message listening infrastructure, focusing on KafkaMessageListenerContainer and ConcurrentMessageListenerContainer , and explains how they are managed by Spring’s lifecycle.

KafkaMessageListenerContainer

Provided by Spring, it listens to and pulls messages, converting them before invoking methods annotated with @KafkaListener. The entry point is doStart(), which creates a ListenerConsumer (a background thread implementing Runnable) and submits it to a thread pool.

protected void doStart() {
    if (isRunning()) {
        return;
    }
    if (this.clientIdSuffix == null) { // stand‑alone container
        checkTopics();
    }
    ContainerProperties containerProperties = getContainerProperties();
    checkAckMode(containerProperties);
    // ...
    // 创建ListenerConsumer消费者并放入到线程池中执行
    this.listenerConsumer = new ListenerConsumer(listener, listenerType);
    setRunning(true);
    this.startLatch = new CountDownLatch(1);
    this.listenerConsumerFuture = containerProperties
        .getConsumerTaskExecutor()
        .submitListenable(this.listenerConsumer);
    // ...
}

ListenerConsumer#run

The consumer thread continuously polls and processes messages, handling wake‑up exceptions and wrapping up on termination.

public void run() { // NOSONAR complexity
    // ...
    Throwable exitThrowable = null;
    while (isRunning()) {
        try {
            // 拉去消息并处理消息
            pollAndInvoke();
        } catch ( @SuppressWarnings UNUSUED WakeupException e) {
            // ...
        }
        // ...
    }
    wrapUp(exitThrowable);
}

ConcurrentMessageListenerContainer

This container creates multiple KafkaMessageListenerContainer instances based on the configured concurrency, enabling parallel consumption.

protected void doStart() {
    if (!isRunning()) {
        checkTopics();
        // ...
        setRunning(true);
        for (int i = 0; i < this.concurrency; i++) {
            KafkaMessageListenerContainer<K, V> container =
                constructContainer(containerProperties, topicPartitions, i);
            String beanName = getBeanName();
            container.setBeanName((beanName != null ? beanName : "consumer") + "-" + i);
            // ...
            if (isPaused()) {
                container.pause();
            }
            // 这里调用KafkaMessageListenerContainer启动相关监听方法
            container.start();
            this.containers.add(container);
        }
    }
}

@KafkaListener Bridge

The @KafkaListener annotation links business logic to the listener containers. KafkaListenerAnnotationBeanPostProcessor scans for this annotation, creates a container via KafkaListenerContainerFactory, and starts it.

Spring Boot Auto‑Configuration for Kafka

Spring Boot automatically provides beans such as KafkaTemplate, ConsumerFactory, and ProducerFactory when they are missing. Two key configuration classes are:

KafkaAutoConfiguration : Generates default Kafka beans.

KafkaAnnotationDrivenConfiguration : Handles annotation‑driven listeners like @KafkaListener, creating a default kafkaListenerContainerFactory if none is defined.

Single‑Message Processing Example

@Configuration
@EnableKafka
public class KafkaConfig {
    @Bean
    KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setConcurrency(3);
        factory.getContainerProperties().setPollTimeout(3000);
        return factory;
    }

    @Bean
    public ConsumerFactory<Integer, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString());
        // ... other properties ...
        return props;
    }
}

@KafkaListener(id = "myListener", topics = "myTopic", autoStartup = "${listen.auto.start:true}", concurrency = "${listen.concurrency:3}")
public void listen(String data) {
    // ... handle single message ...
}

Batch Processing Example

@Configuration
@EnableKafka
public class KafkaConfig {
    @Bean
    public KafkaListenerContainerFactory<?, ?> batchFactory() {
        ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        factory.setBatchListener(true); // enable batch
        return factory;
    }

    @Bean
    public ConsumerFactory<Integer, String> consumerFactory() {
        return new DefaultKafkaConsumerFactory<>(consumerConfigs());
    }

    @Bean
    public Map<String, Object> consumerConfigs() {
        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString());
        // ... other properties ...
        return props;
    }
}

@KafkaListener(id = "list", topics = "myTopic", containerFactory = "batchFactory")
public void listen(List<String> list) {
    // ... handle batch of messages ...
}

Key Points and Best Practices

If a custom ContainerFactory bean name differs from kafkaListenerContainerFactory, Spring creates a new instance via KafkaAnnotationDrivenConfiguration. Ensure the correct factory is referenced in @KafkaListener.

Single‑message and batch factories can coexist; the default bean name is used unless explicitly specified.

Spring’s Kafka module abstracts the low‑level client, allowing developers to focus on business logic while Spring manages lifecycle, threading, and error handling.

Conclusion

Spring integrates Kafka through the spring‑kafka module, providing convenient abstractions such as KafkaMessageListenerContainer and the @KafkaListener annotation. These tools enable both single‑message and batch processing with minimal configuration, while still allowing advanced customizations when needed.