def _abort_batches(self, error):
        """Go through incomplete batches and abort them."""
        for batch in self._incomplete.all():
            tp = batch.topic_partition
            # Close the batch before aborting
            with self._tp_lock(tp):
                batch.records.close()
                self._batches[tp].remove(batch)
            batch.abort(error)
            self.deallocate(batch)
Go through incomplete batches and abort them.
kafka/producer/record_accumulator.py