def abort(self, exception):
        """Abort the batch and complete the future and callbacks."""
        if self._final_state is not None:
            raise Errors.IllegalStateError("Batch has already been completed in final state: %s" % self._final_state)
        self._final_state = FinalState.ABORTED

        log.debug("Aborting batch for partition %s: %s", self.topic_partition, exception)
        self._complete_future(-1, -1, exception)
Abort the batch and complete the future and callbacks.
kafka/producer/record_accumulator.py