org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message

Question

I am new to Apache Kafka and am able to send the message(in JSON format) from the sender, but not able to consume at consumer service.

This is my Code:

Sender Service

@Service
public class SenderService {
    private static final Logger LOG = LoggerFactory.getLogger(SenderService.class);
    
    @Autowired
    private KafkaTemplate<String, IdName> kafkaTemplate;
    
    @Value("${app.topic.email}")
    private String topic;
    
    public void send(IdName idName) {
        LOG.info("Sending Data='{}' to topic='{}' ", idName, topic);
        
        Message<IdName> message = MessageBuilder.withPayload(idName).setHeader(KafkaHeaders.TOPIC, topic)
            .setHeader(KafkaHeaders.MESSAGE_KEY, "TestMessage")
            .build();
        kafkaTemplate.send(message);
    }
}

###Consumer Service

@Service
public class ConsumerService {
    private static final Logger LOG = LoggerFactory.getLogger(ConsumerService.class);
    
    @Autowired
    MailSender mailSender;
    
    @KafkaListener(topics = "${app.topic.email}")
    public void receive(@Payload IdName data,
        @Header MessageHeaders headers) throws Exception{
        LOG.info("Received data='{}'", data);
    }
 }

I receive the below exception

2018-09-21 11:01:41.738 ERROR 63487 --- [ntainer#0-0-C-1] o.s.kafka.listener.LoggingErrorHandler   : Error while processing: ConsumerRecord(topic = emailclient, partition = 0, offset = 4, CreateTime = 1537507901660, serialized key size = 11, serialized value size = 122, headers = RecordHeaders(headers = [RecordHeader(key = __TypeId__, value = [99, 111, 109, 46, 97, 98, 99, 112, 108, 117, 115, 100, 46, 97, 112, 112, 108, 105, 99, 97, 116, 105, 111, 110, 46, 100, 111, 109, 97, 105, 110, 46, 73, 100, 78, 97, 109, 101])], isReadOnly = false), key = TestMessage, value = com.*****.****.domain.IdName@4a2d3006)

org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Endpoint handler details:
Method [public void com.*****.****.service.ConsumerService.receive(com.*****.****.domain.IdName,org.springframework.messaging.MessageHeaders) throws java.lang.Exception]

Bean [com.*****.****.service.ConsumerService@4649d70a]; nested exception is org.springframework.messaging.MessageHandlingException: Missing header 'headers' for method parameter type [class org.springframework.messaging.MessageHeaders], failedMessage=GenericMessage [payload=com.*****.****.domain.IdName@4a2d3006, headers={kafka_offset=4, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@53fdf4bb, kafka_timestampType=CREATE_TIME, kafka_receivedMessageKey=TestMessage, kafka_receivedPartitionId=0, kafka_receivedTopic=emailclient, kafka_receivedTimestamp=1537507901660, __TypeId__=[B@707343c2}]

Can anybody help me out?

Michael Altenburger Michael Altenburger · Accepted Answer · 2018-09-21T05:48:43

If you want to have access to all headers in the listener, you are using the wrong annotation, it should be @Headers instead of @Header. The resulting code looks like this:

@Service
public class ConsumerService {
private static final Logger LOG = LoggerFactory.getLogger(ConsumerService.class);

    @Autowired
    MailSender mailSender;

    @KafkaListener(topics = "${app.topic.email}")
    public void receive(@Payload IdName data,
        @Headers MessageHeaders headers) throws Exception{
        LOG.info("Received data='{}'", data);
    }
 }

You could of course also only inject the message key, like so:

@Service
public class ConsumerService {
private static final Logger LOG = LoggerFactory.getLogger(ConsumerService.class);

    @Autowired
    MailSender mailSender;

    @KafkaListener(topics = "${app.topic.email}")
    public void receive(@Payload IdName data,
        @Header(KafkaHeaders.MESSAGE_KEY) String messageKey) throws Exception{
        LOG.info("Received data='{}'", data);
    }
 }

org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message

This is my Code:

Sender Service

1 Answers