'Go through incomplete batches and abort them.'

def _abort_batches(self):

    error = Errors.IllegalStateError('Producer    is    closed    forcefully.')
    for batch in self._incomplete.all():
        tp = batch.topic_partition
        with self._tp_locks[tp]:
            batch.records.close()
        batch.done(exception=error)
        self.deallocate(batch)
