title: Kafka consumer behaviour on failed deserialization
body: <p>I was wondering how to tell kafka consumer when Deserialization of an object fails
  to drop the record from kafka, log the error and continue listening to other incoming messages.</p>
  
  <p>I've got my consumer configured as below:
  ConsumerProperties:</p>
  
  <pre><code>Map&lt;String, Object&gt; consumerProperties = new HashMap&lt;&gt;();
      consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
              kafkaPropertiesConfiguration.getBootstrapServers());
      consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
              StringDeserializer.class.getName());
  
      consumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
              NullDeserializer.class.getName());
  
      consumerProperties.put(ConsumerConfig.CLIENT_ID_CONFIG,
              kafkaTelecontrolEventConsumerProperties.getClientIdConfig());
      consumerProperties.put(ConsumerConfig.GROUP_ID_CONFIG,
              kafkaTelecontrolEventConsumerProperties.getGroupIdConfig());
  
      return new DefaultKafkaConsumerFactory&lt;&gt;(consumerProperties); 
  </code></pre>
  
  <p>NullDeserializer.class is my deserializer used for testing:</p>
  
  <pre><code>@Log4j2
  public class NullDeserializer implements Deserializer&lt;Object&gt; {
      @Override
      public void configure(Map&lt;String, ?&gt; configs, boolean isKey) {
  
      }
  
      @Override
      public Object deserialize(String topic, byte[] data) {
          try {
              TeleControl.Event.parseFrom(data) ;
          } catch (InvalidProtocolBufferException e) {
              throw new RuntimeException(e);
          }
          return null;
      }
  
      @Override
      public void close() {
  
      }
  }
  </code></pre>
  
  <p>IntegrationFlow:</p>
  
  <pre><code>return IntegrationFlows
                  .from(Kafka
                          .messageDrivenChannelAdapter(telecontrolEventConsumerFactory,
                                  eventConsumerProperties.getTelecontrolEventTopic())
                  )
                  .handle(System.err::println)
                  .get();
  </code></pre>
  
  <p>My main problem is when new RuntimeException is thrown when parsing fails, how to tell kafkaConsumer log the error, drop the record and continue processing next kafka message.</p>