def run(self):
        """The main run loop for the sender thread."""
        log.debug("%s: Starting Kafka producer I/O thread.", str(self))

        # main loop, runs until close is called
        while self._running:
            try:
                self.run_once()
            except Exception:
                log.exception("%s: Uncaught error in kafka producer I/O thread", str(self))

        log.debug("%s: Beginning shutdown of Kafka producer I/O thread, sending"
                  " remaining records.", str(self))

        # okay we stopped accepting requests but there may still be
        # requests in the accumulator or waiting for acknowledgment,
        # wait until these are completed.
        while (not self._force_close
               and (self._accumulator.has_undrained()
                    or self._client.in_flight_request_count() > 0)):
            try:
                self.run_once()
            except Exception:
                log.exception("%s: Uncaught error in kafka producer I/O thread", str(self))

        if self._force_close:
            # We need to fail all the incomplete batches and wake up the
            # threads waiting on the futures.
            self._accumulator.abort_incomplete_batches()

        try:
            self._client.close()
        except Exception:
            log.exception("%s: Failed to close network client", str(self))

        log.debug("%s: Shutdown of Kafka producer I/O thread has completed.", str(self))
The main run loop for the sender thread.
kafka/producer/sender.py