title: How to handle SerializationException after deserialization
body: <p>I am using Avro and Schema registry with my Spring Kafka setup.</p>
  
  <p>I would like to somehow handle the <code>SerializationException</code>, which might be thrown during deserialization.</p>
  
  <p>I found the following two resource:</p>
  
  <p><a href="https://github.com/spring-projects/spring-kafka/issues/164" rel="noreferrer">https://github.com/spring-projects/spring-kafka/issues/164</a></p>
  
  <p><a href="https://stackoverflow.com/questions/45584504/how-do-i-configure-spring-kafka-to-ignore-messages-in-the-wrong-format">How do I configure spring-kafka to ignore messages in the wrong format?</a></p>
  
  <p>These resources suggest that I return null instead of throwing an <code>SerializationException</code> when deserializing and listen for <code>KafkaNull</code>. This solution works just fine.</p>
  
  <p>I would however like to be able to throw an exception instead of returning null. </p>
  
  <p><a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-161%3A+streams+deserialization+exception+handlers" rel="noreferrer">KIP-161</a> and <a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-210+-+Provide+for+custom+error+handling++when+Kafka+Streams+fails+to+produce" rel="noreferrer">KIP-210</a> provide better features to handling exceptions. I did find some resources mentioning KIP-161 in Spring Cloud, but nothing specific about Spring-Kafka.</p>
  
  <p>Does anyone know how to catch <code>SerializationException</code> in Spring Boot?</p>
  
  <p>I am using Spring Boot 2.0.2</p>
  
  <p>Edit: I found a solution.</p>
  
  <p>I would rather throw an exception and catch it than having to return null or <code>KafkaNull</code>. I am using my custom Avro serializer and deserializer in multiple different project, some of which are not Spring. If I changed my Avro serializer and deserializer then some of the other projects would need to be changed to expect the deserializer to return null.</p>
  
  <p>I would like to shutdown the container, such that I do not lose any messages. The SerializationException should never be expected in production. The SerializationException should only be able to happen if Schema Registry is down or if an unformatted message somehow is sent to the production kafka. Either way, SerializationException should only happen very rarely, and if it happens then I want to shutdown the container such that no messages are lost and I can investigate the issue.</p>
  
  <p>Just take into consideration that will catch all exceptions from your consumer container. In my specific case I just want to only shutdown if it is a <code>SerializationException</code></p>
  
  <pre><code>public class SerializationExceptionHandler extends ContainerStoppingErrorHandler {
  
      @Override
      public void handle(Exception thrownException, List&lt;ConsumerRecord&lt;?, ?&gt;&gt; records, Consumer&lt;?, ?&gt; consumer,
                         MessageListenerContainer container) {
  
          //Only call super if the exception is SerializationException
          if (thrownException instanceof SerializationException) {
              //This will shutdown the container.
              super.handle(thrownException, records, consumer, container);
          } else {
              //Wrap and re-throw the exception
              throw new KafkaException("Kafka Consumer Container Error", thrownException);
          }
      }
  }
  </code></pre>
  
  <p>This handler is passed to the consumer container. Below is an example of a 
  <code>KafkaListenerContainerFactory</code> bean.</p>
  
  <pre><code>@Bean
  public KafkaListenerContainerFactory&lt;ConcurrentMessageListenerContainer&lt;Integer, String&gt;&gt;
  kafkaListenerContainerFactory(JpaTransactionManager jpa, KafkaTransactionManager&lt;?, ?&gt; kafka) {
  
      ConcurrentKafkaListenerContainerFactory&lt;Integer, String&gt; factory =
              new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
  
      factory.setConsumerFactory(consumerFactory());
      factory.setConcurrency(1);
      factory.getContainerProperties().setPollTimeout(3000);
  
      factory.getContainerProperties().setErrorHandler(new SerializationExceptionHandler());
  
      factory.getContainerProperties().setTransactionManager(chainedTxM(jpa, kafka));
      return factory;
  }
  </code></pre>