Kafka消费者@KafkaListener标注的consume()方法未触发排查

编辑更新：问题根因为代码引用了不存在的配置属性路径，调整Consumer.java类上的注解配置后，消费者逻辑已可正常运行，修正后的注解如下：

@ConditionalOnProperty(value = "aaronshaver.kafka.consumer-enabled", havingValue = "true")

问题描述 #

我搭建了一套简单的Kafka环境，已确认生产者可正常生产消息，通过如下SendMessageTask.java中的代码可验证发送结果：

ListenableFuture<SendResult<String, String>> listenableFuture = this.producer.sendMessage("INPUT_DATA", "IN_KEY", LocalDate.now().toString());

SendResult<String, String> result = listenableFuture.get();
logger.info(String.format("\nProduced:\ntopic: %s\noffset: %d\npartition: %d\nvalue size: %d\n", 
    result.getRecordMetadata().topic(), 
    result.getRecordMetadata().offset(), 
    result.getRecordMetadata().partition(), 
    result.getRecordMetadata().serializedValueSize()
));

控制台可正常输出包含发送结果的消息内容。

我始终无法明确消费者的Kafka监听方法为何未被调用，以下是我使用的消费者代码：
Consumer.java

@Service
@ConditionalOnProperty(value = "example.kafka.consumer-enabled", havingValue = "true")
public class Consumer {

    private final Logger logger = LoggerFactory.getLogger(Producer.class);

    @KafkaListener(topics = "INPUT_DATA", groupId = "fooGroup")
    public void listenGroupFoo(String message) {
        System.out.println("Received Message in group fooGroup: " + message);
    }

    @KafkaListener(topics = {"INPUT_DATA"})
    public void consume(
            final @Payload String message,
            final @Header(KafkaHeaders.OFFSET) Integer offset,
            final @Header(KafkaHeaders.RECEIVED_MESSAGE_KEY) String key,
            final @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition,
            final @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
            final @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long ts,
            final Acknowledgment acknowledgment
    ) {
        logger.info(String.format("#### -> Consumed message -> TIMESTAMP: %d\n%s\noffset: %d\nkey: %s\npartition: %d\ntopic: %s", 
            ts, message, offset, key, partition, topic
        ));
        acknowledgment.acknowledge();
    }
}

同时编写了如下消费端配置类：
KafkaConsumerConfig.java

@EnableKafka
@Configuration
public class KafkaConsumerConfig {

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {
        Map<String, Object> props = new HashMap<>();
        props.put(
                ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
                "localhost:9092");
        props.put(
                ConsumerConfig.GROUP_ID_CONFIG,
                "fooGroup");
        props.put(
                ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class);
        props.put(
                ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                StringDeserializer.class);
        return new DefaultKafkaConsumerFactory<>(props);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory());
        return factory;
    }
}

原以为标注了@KafkaListener的方法会被自动调用，但实际并未触发，不确定是对Kafka监听运行机制存在理解偏差，还是配置存在问题，寻求相关排查思路。

以下是src/main/resources/application.yml中配置的Spring与Kafka相关属性：

aaronshaver:
  kafka:
    consumer-enabled: ${consumer-enabled:true}
spring:
  kafka:
    bootstrap-servers: ${kafka_bootstrap_servers:localhost:29092}
    properties:
      sasl:
        jaas:
          config: org.apache.kafka.common.security.plain.PlainLoginModule required username=${kafka_username:'admin'} password=${kafka_password:'admin-secret'};
        mechanism: PLAIN
      security:
        protocol: SASL_PLAINTEXT
      consumer:
        auto-offset-reset: earliest
        group-id: aaronshaver
        key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
        value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
        max-poll-records: 1
        fetch-max-wait: 36000
        enable-auto-commit: false
        client-id: aaronshaver
      producer:
        client-id: aaronshaver
        key-serializer: org.apache.kafka.common.serializatoin.StringSerializer
        value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
        retries: 2
      jaas:
        enabled: true
      listener:
        poll-timeout: 1800000
        concurrency: 1
        ack-mode: manual_immediate

排查修正方案 #

逐一对齐代码和配置，核心问题点如下：

• 条件注解配置路径不匹配：Consumer类上的@ConditionalOnProperty指定的属性路径example.kafka.consumer-enabled在配置文件中完全不存在，Spring判定条件不满足，直接跳过该类的容器初始化，类内的@KafkaListener方法自然不会被扫描。将注解的属性路径调整为配置中实际存在的aaronshaver.kafka.consumer-enabled即可。
• YAML配置缩进错误：原配置中spring节点错误嵌套在aaronshaver节点下，会导致Spring Boot原生Kafka自动配置完全失效，需要将spring节点提升到根层级，和aaronshaver节点同级。
• 自定义消费配置参数不一致：手写的KafkaConsumerConfig中硬编码了Kafka集群地址localhost:9092、组IDfooGroup，且没有配置SASL认证参数，和实际配置的集群地址localhost:29092、SASL_PLAINTEXT认证机制完全不匹配，就算类被初始化，消费者也无法正常连接集群。可直接删除该自定义配置类，复用Spring Boot自动装配的Kafka连接能力，避免参数双重维护不一致。
• 生产端配置拼写错误：spring.kafka.producer节点下的序列化配置存在两处错误：一是key-serializer的类名包路径拼写错误（serializatoin），二是错误将value-serializer配置写成了value-deserializer，会导致生产者初始化异常，与已验证的发送成功状态矛盾，需要修正为正确的序列化类路径。

内容的提问来源于stack exchange，提问作者Aaron