def _complete_batch(self, batch, error, base_offset, timestamp_ms=None):
        """Complete or retry the given batch of records.

        Arguments:
            batch (ProducerBatch): 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:
            if self._can_retry(batch, error):
                # retry
                log.warning("%s: Got error produce response on topic-partition %s,"
                            " retrying (%s attempts left). Error: %s",
                            str(self), batch.topic_partition,
                            self.config['retries'] - batch.attempts - 1,
                            error)

                # If idempotence is enabled only retry the request if the batch matches our current producer id and epoch
                if not self._transaction_manager or self._transaction_manager.producer_id_and_epoch.match(batch):
                    log.debug("%s: Retrying batch to topic-partition %s. Sequence number: %s",
                              str(self), batch.topic_partition,
                              self._transaction_manager.sequence_number(batch.topic_partition) if self._transaction_manager else None)
                    self._accumulator.reenqueue(batch)
                    self._maybe_remove_from_inflight_batches(batch)
                    if self._sensors:
                        self._sensors.record_retries(batch.topic_partition.topic, batch.record_count)
                else:
                    log.warning("%s: Attempted to retry sending a batch but the producer id/epoch changed from %s/%s to %s/%s. This batch will be dropped",
                                str(self), batch.producer_id, batch.producer_epoch,
                                self._transaction_manager.producer_id_and_epoch.producer_id,
                                self._transaction_manager.producer_id_and_epoch.epoch)
                    self._fail_batch(batch, error, base_offset=base_offset, timestamp_ms=timestamp_ms)
            else:
                if error is Errors.TopicAuthorizationFailedError:
                    error = error(batch.topic_partition.topic)

                # tell the user the result of their request
                self._fail_batch(batch, error, base_offset=base_offset, timestamp_ms=timestamp_ms)

            if error is Errors.UnknownTopicOrPartitionError:
                log.warning("%s: Received unknown topic or partition error in produce request on partition %s."
                            " The topic/partition may not exist or the user may not have Describe access to it",
                            str(self), batch.topic_partition)

            if getattr(error, 'invalid_metadata', False):
                self._metadata.request_update()

        else:
            if batch.done(base_offset=base_offset, timestamp_ms=timestamp_ms):
                self._maybe_remove_from_inflight_batches(batch)
                self._accumulator.deallocate(batch)

            if self._transaction_manager and self._transaction_manager.producer_id_and_epoch.match(batch):
                self._transaction_manager.increment_sequence_number(batch.topic_partition, batch.record_count)
                log.debug("%s: Incremented sequence number for topic-partition %s to %s", str(self), batch.topic_partition,
                          self._transaction_manager.sequence_number(batch.topic_partition))

        # Unmute the completed partition.
        if self.config['guarantee_message_order']:
            self._accumulator.muted.remove(batch.topic_partition)
Complete or retry the given batch of records.

        Arguments:
            batch (ProducerBatch): 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
        
kafka/producer/sender.py