CODEWITHSUNDEEP
javaspringapache-kafkaspring-kafka
Matt
Matt

Reputation: 616

Spring kafka @KafkaListener is not being invoked

I'm trying to integrate kafka with my Spring boot (v2.0.6.RELEASE) app using spring-kafka. For now I want to have a consumer and a producer. I got the producer to work just fine, I can see the messages being sent to the topic through a console consumer. I can't get the consumer code to work, it does not get invoked when a new message appears in the kafka topic.

Here is my kafka configuration class:


@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());
    }

}

Here is my pom dependency:

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

And the consumer code:


@Slf4j
public class MyMessageConsumer {

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

I am running kafka on my computer and as I said - I can see the messages appearing in the console consumer. What could be the reason that my Spring based consumer won't receive messages?

EDIT: Here is the deserializer class:


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() {

    }
}

Upvotes: 5

Views: 14262

Answers (1)

Abdelghani Roussi
Abdelghani Roussi

Reputation: 2817

Because your class MyMessageConsumer is not scanned by spring due to missing of stereotype annotation, so adding a @Component for example to this class will resolve your issue : 

@Slf4j
@Component
public class MyMessageConsumer {

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

I think that it's a good idea to specify a consumer group in your @KafkaListner annotation

Upvotes: 12

Related Questions