Spring kafka @KafkaListener не вызывается

Question

Я пытаюсь интегрировать kafka с моим приложением Spring boot (v2.0.6.RELEASE), используя spring-kafka. Сейчас я хочу иметь потребителя и производителя. Я заставил продюсера работать нормально, я вижу сообщения, отправляемые в тему через консольного потребителя. Я не могу заставить работать потребительский код, он не вызывается, когда в теме kafka появляется новое сообщение.

Вот мой класс конфигурации kafka:

@Configuration
@EnableKafka
public class KafkaConfiguration {

    @Bean
    public Map<String, Object> producerConfigs() {
        Map<String, Object> props = new HashMap<>();

        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, <path to my custom serializer>);
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");

        return props;
    }

    @Bean
    public Map<String, Object> consumerConfigs(){
        Map<String, Object> props = new HashMap<>();

        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, <path to my custom deserializer>);
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
//        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        return props;
    }

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

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

    @Bean
    public ProducerFactory<String, List<MyMessage>> producerFactory() {
        return new DefaultKafkaProducerFactory<>(producerConfigs());
    }

    @Bean
    public KafkaTemplate<String, List<MyMessage>> kafkaTemplate() {
        return new KafkaTemplate<>(producerFactory());
    }

}

Вот моя зависимость от пом:

<dependency>
  <groupId>org.springframework.kafka</groupId>
  <artifactId>spring-kafka</artifactId>
</dependency>

И потребительский код:

@Slf4j
public class MyMessageConsumer {

    @KafkaListener(topicPattern = "test_topic")
    public void receive(List<MyMessage> myMessages){
        log.info("Received payload {}", myMessages);
    }
}

Я запускаю kafka на своем компьютере и, как я уже сказал, я вижу сообщения, появляющиеся в консоли потребителя. В чем может быть причина того, что мой потребитель на основе Spring не будет получать сообщения?

EDIT: Вот класс десериализатора:

public class TagEventDeserializer implements Deserializer<List<MyMessage>> {
    @Override
    public void configure(Map map, boolean b) {

    }

    @Override
    public List<MyMessage> deserialize(String s, byte[] bytes) {
        ObjectMapper objectMapper = new ObjectMapper();
        List<MyMessage> myMessages = null;
        try{
            myMessages = objectMapper.readValue(bytes , List.class);
        } catch (Exception e){
            e.printStackTrace();
        }
        return myMessages;
    }

    @Override
    public void close() {

    }
}

Answer 1

Поскольку ваш класс MyMessageConsumer не сканируется весной из-за отсутствия аннотации стереотипа, поэтому добавление @Component, например, к этому классу, решит вашу проблему:

@Slf4j
@Component
public class MyMessageConsumer {

    @KafkaListener(topicPattern = "test_topic")
    public void receive(List<MyMessage> myMessages){
        log.info("Received payload {}", myMessages);
    }
}

Я думаю, что это хорошая идея, чтобы указать группу потребителей в вашей @KafkaListner аннотации