Critical Description
KAFKA-13348 added the ability to ignore producer exceptions by setting error.tolerance to all. When this is set to all a null record metadata is passed to commitRecord() and the task continues.
The issue is that records are tracked by SubmittedRecords and the first time an error happens the code does not ack the record with the error and just skips it so it will not have the offsets committed or be removed from SubmittedRecords before calling commitRecord().
This leads to a bug where future offsets won't be committed anymore and also a memory leak because the algorithm that removes acked records from the internal map to commit offsets looks at the head of the Deque where the records are tracked in and if it sees the record is unacked it will not process anymore removals. This leads to all new records that go through the task to continue to be added and not have offsets committed and never removed from tracking until an OOM error occurs.
The fix is to make sure to ack the failed records so they can have their offsets commited and be removed from tracking. This is fine to do as the records are intended to be skipped and not reprocessed. Metrics also need to be updated as well.