docs: def _complete_batch(self, batch, error, base_offset, timestamp_ms=None):
          """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
          """
          # Standardize no-error to None
          if error is Errors.NoError:
              error = None
  
          if error is not None and self._can_retry(batch, error):
              # retry
              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)
  
              # tell the user the result of their request
              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()
  
          # Unmute the completed partition.
          if self.config['guarantee_message_order']:
              self._accumulator.muted.remove(batch.topic_partition)