[BUG] Kafka source stops committing offsets after consuming a record that results in deserialization error

[hshardeesi]

Describe the bug
With acknowledgments enabled, if KafkaConsumer encounters deserialization error while consuming a record then kafka source plugin stops committing offsets for that partition. Subsequent records are consumed and pushed to sink, but committed offset remains stuck on the offset one before the record that encountered deserialization exception.

To Reproduce

• Create kafka topic and ingest json records.
• Create pipeline with kafka source and "json" as serde_format
• Send one text (non-json) record to the topic.
• Deserialization exception will be thrown in pipeline.
• Describe kafka consumer group on kafka cluster to observe partition LAG/committed offsets.
• Any further records sent to the error partition will not result in new offsets being committed and LAG will remain non-zero.

Expected behavior
Failed records should not result in stuck committed offsets, commits should resume after accounting error records in NumberOfDeserializationErrors metric. (In future, failed records can be sent to source/pipeline level dlq if available)