'Complete or retry the given batch of records.
Arguments:
batch (RecordBatch): The record batch
error (Exception): The error (or None if none)
base_offset (int): The base offset assigned to the records if successful
timestamp_ms (int, optional): The timestamp returned by the broker for this batch'

def _complete_batch(self, batch, error, base_offset, timestamp_ms=None):

    if (error is Errors.NoError):
        error = None
    if ((error is not None) and self._can_retry(batch, error)):
        log.warning('Got    error    produce    response    on    topic-partition    %s,    retrying    (%d    attempts    left).    Error:    %s', batch.topic_partition, ((self.config['retries'] - batch.attempts) - 1), error)
        self._accumulator.reenqueue(batch)
        self._sensors.record_retries(batch.topic_partition.topic, batch.record_count)
    else:
        if (error is Errors.TopicAuthorizationFailedError):
            error = error(batch.topic_partition.topic)
        batch.done(base_offset, timestamp_ms, error)
        self._accumulator.deallocate(batch)
        if (error is not None):
            self._sensors.record_errors(batch.topic_partition.topic, batch.record_count)
    if getattr(error, 'invalid_metadata', False):
        self._metadata.request_update()
    if self.config['guarantee_message_order']:
        self._accumulator.muted.remove(batch.topic_partition)
