title: Spring-kafka error handling with DeadLetterPublishingRecoverer
body: <p>I am trying to implement error handling in Spring boot kafa. In my Kafka listener I am throwing a runtime exception as per below:</p>
  
  <pre><code>@KafkaListener(topics= "Kafka-springboot-example", groupId="group-employee-json")
      public void consumeEmployeeJson(Employee employee) {
          logger.info("Consumed Employee JSON: "+ employee);
  
          if(null==employee.getEmployeeId()) {
              throw new RuntimeException("failed");
              //throw new ListenerExecutionFailedException("failed");
          }
      }
  </code></pre>
  
  <p>And I have configured error handling as per below:</p>
  
  <pre><code>@Configuration
  @EnableKafka
  public class KafkaConfiguration {
  
      @Bean
      public ConcurrentKafkaListenerContainerFactory&lt;Object, Object&gt; containerFactory(
              ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
              ConsumerFactory&lt;Object, Object&gt; kafkaConsumerFactory,
              KafkaTemplate&lt;Object, Object&gt; template){
  
          ConcurrentKafkaListenerContainerFactory&lt;Object, Object&gt; factory= new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          configurer.configure(factory, kafkaConsumerFactory);
          factory.setErrorHandler(new SeekToCurrentErrorHandler(
                  new DeadLetterPublishingRecoverer(template)));
  
          return factory;
  
      }
  }
  </code></pre>
  
  <p>And my listener for DLT is as per below:</p>
  
  <pre><code>@KafkaListener(topics= "Kafka-springboot-example.DLT", groupId="group-employee-json")
      public void consumeEmployeeErrorJson(Employee employee) {
          logger.info("Consumed Employee JSON frpm DLT topic: "+ employee);
      }
  </code></pre>
  
  <p>But my message is not getting published to DLT topic.</p>
  
  <p>Any idea what I am doing wrong?</p>
  
  <p>Edited:</p>
  
  <p>application.properties</p>
  
  <pre><code>server.port=8088
  
  #kafka-producer-config
  spring.kafka.producer.bootstrap-servers=localhost:9092
  spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
  spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer
  
  
  #Kafka consumer properties
  spring.kafka.consumer.bootstrap-servers=localhost:9092
  spring.kafka.consumer.group-id=group-employee-json
  spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
  spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
  spring.kafka.consumer.properties.spring.json.trusted.packages=*
  </code></pre>