title: org.springframework.kafka.listener.ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
body: <p>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.</p>
  <h3>This is my Code:</h3>
  <h3>Sender Service</h3>
  <pre><code>@Service
  public class SenderService {
      private static final Logger LOG = LoggerFactory.getLogger(SenderService.class);
      
      @Autowired
      private KafkaTemplate&lt;String, IdName&gt; kafkaTemplate;
      
      @Value(&quot;${app.topic.email}&quot;)
      private String topic;
      
      public void send(IdName idName) {
          LOG.info(&quot;Sending Data='{}' to topic='{}' &quot;, idName, topic);
          
          Message&lt;IdName&gt; message = MessageBuilder.withPayload(idName).setHeader(KafkaHeaders.TOPIC, topic)
              .setHeader(KafkaHeaders.MESSAGE_KEY, &quot;TestMessage&quot;)
              .build();
          kafkaTemplate.send(message);
      }
  }
  
  ###Consumer Service
  
  @Service
  public class ConsumerService {
      private static final Logger LOG = LoggerFactory.getLogger(ConsumerService.class);
      
      @Autowired
      MailSender mailSender;
      
      @KafkaListener(topics = &quot;${app.topic.email}&quot;)
      public void receive(@Payload IdName data,
          @Header MessageHeaders headers) throws Exception{
          LOG.info(&quot;Received data='{}'&quot;, data);
      }
   }
  </code></pre>
  <p>I receive the below exception</p>
  <pre><code>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}]
  </code></pre>
  <p>Can anybody help me out?</p>