title: How to handle ValueDeserializationError in confluent_kafka python?
body: <p>This is the base consumer class I'm using for creating new consumers. It works fine for <code>&quot;enable.auto.commit&quot;:True</code> consumer. But when I create a consumer with <code>enable.auto.commit=False</code> and any of the (KeyDeserializationError, ValueDeserializationError) exceptions occurs then I need to manually commit that message in except block. As this base class will be used for auto-commit=True as well, so this line self.consumer.commit() is getting called to these types of consumers also.</p>
  <ol>
  <li>By calling <strong>commit()</strong> for <code>auto.commit=True</code> consumers give any issue internally? (it seems fine when I tried locally)</li>
  <li>What should be ideal handling for (<code>KeyDeserializationError</code>, <code>ValueDeserializationError</code>) exceptions for <code>auto.commit=False</code>?</li>
  </ol>
  <pre class="lang-py prettyprint-override"><code>class KafkaConsumer(object):
      &quot;&quot;&quot;Wrapper over Kafka Consumer&quot;&quot;&quot;
      def __init__(self, topics: list[str], **kwargs: Any):
          config = {
              **kwargs,
          }
          self.consumer = DeserializingConsumer(config)
          self.consumer.subscribe(topics=topics)
  
      def consume(self, poll_timeout_secs: float = 1.0):
          try:
              while True:
                  try:
                      msg = self.consumer.poll(timeout=poll_timeout_secs)
                  except (KeyDeserializationError, ValueDeserializationError) as err:
                      self.consumer.commit()
                      continue
                  if msg is None:
                      continue
                  if msg.error():
                      raise KafkaException(msg.error())
                  else:
                      yield msg
          except:
              self.consumer.close()
  
  # create consumer object auto.commit=True/False
  
  kafka_consumer = KafkaConsumer(topics=topics, **kwargs) # i can pass &quot;enable.auto.commit&quot;:False for manual commit mode.
  
  # Actual consuming business logic
  
  for message in kafka_consumer.consume():
      try:
          event = message.value()
          logger.info(f&quot;message {event}&quot;)
      except Exception as e:
          logger.exception(f'Unable to consume data from kafka {e}')
  finally:
  pass
  # kafka_consumer.consumer.commit(message=message) # in case of manual commit consumer mode
  </code></pre>