KAFKA-15800: Prevent DataExceptions from corrupting KafkaOffsetBackingStore

[gharris1727]

https://issues.apache.org/jira/browse/KAFKA-15800

The JsonConverter may throw DataException when encountering malformed data. This causes part or all of the batch of ConsumerRecords from the KafkaBasedLog consumer to be dropped, corrupting the state of the KafkaOffsetBackingStore.

This is a narrow-scope fix for just the KafkaOffsetBackingStore, and i'd like to follow up with a fix to the KafkaBasedLog to make these sorts of mistakes less impactful in the future.

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[yashmayya]

Thanks for the fix, Greg! The change itself makes sense to me, but I was just wondering about the scenarios that would lead to something like this occurring? The offsets that the source connector generates would be serialized in the runtime by the JSON converter itself before being written to the offsets topic. Is this fix for cases where connectors or operators directly produce bad data to the offsets topic for a Connect cluster?

[gharris1727]

Is this fix for cases where connectors or operators directly produce bad data to the offsets topic for a Connect cluster?

I do not know how we got malformed data in the topic exactly, but the invalid records did appear to be almost valid source offsets, they were just missing a double quote in the middle of the key. It may have been from a bad converter version, a bit flip, or something else, I can't be sure.

It may have been an operator mis-typing during a manual remediation, I just don't have direct evidence for that. If that was the case, KIP-875 (which introduced this bug) should also make it less likely in the future.

[yashmayya]

LGTM, thanks Greg!

[gharris1727]

Test failures appear unrelated, and the connect and mirror tests pass locally.