title: Kafka - How to skip a bad message in an offset and consume the rest
body: <p>I am using Kafka with Avro for serialization/de-serialization for processing events. If by chance a bad data that doesnt comply to the avro schema gets in the topic, </p>
  
  <pre><code>.springframework.kafka.listener.KafkaMessageListenerContainer$ListenerConsumer app: host: dcId: envId: url: reqId: jsess: secSessId: logUser: effUser: impUser: channelName: - Container exception
  org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition EventProcessor-0 at offset 2845. If needed, please seek past the record to continue consumption.
  </code></pre>
  
  <p>and the message keeps growing for the same offset. Is there a possibility to just skip this offset and continue to read from the further offsets and if the same happens again, skip that too ? </p>
  
  <p>Consumer code:</p>
  
  <pre><code>@KafkaListener(topics = "EventProcessor", containerFactory = "eventProcessorListenerContainerFactory")
      public void listen(Event payLoad) {
  
          System.out.println("REceived  message ===&gt; " + payLoad);
  
      }
  </code></pre>
  
  <p>Factory :</p>
  
  <pre><code>@Bean
      KafkaListenerContainerFactory&lt;ConcurrentMessageListenerContainer&lt;String, Event&gt;&gt; eventProcessorListenerContainerFactory() {
  
          Map&lt;String, Object&gt; propMap = new HashMap&lt;String, Object&gt;();
          propMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);
          propMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
          propMap.put(ConsumerConfig.GROUP_ID_CONFIG, "EventProcessorConsumer");
  
          DefaultKafkaConsumerFactory&lt;String, Event&gt; consuemrFactory = new DefaultKafkaConsumerFactory&lt;String, Event&gt;(
                  propMap);
  
          consuemrFactory.setValueDeserializer(new AvroDeSerializer&lt;Event&gt;(
                  Event.class));
          ConcurrentKafkaListenerContainerFactory&lt;String, Event&gt; listenerFactory = new ConcurrentKafkaListenerContainerFactory&lt;&gt;();
          listenerFactory.setConsumerFactory(consuemrFactory);
          listenerFactory.setConcurrency(3);
          listenerFactory.setRetryTemplate(retryTemplate());
          listenerFactory.getContainerProperties().setPollTimeout(3000);
          return listenerFactory;
      }
  </code></pre>