Invoking this method makes all buffered records immediately available
        to send (even if linger_ms is greater than 0) and blocks on the
        completion of the requests associated with these records. The
        post-condition of :meth:`~kafka.KafkaProducer.flush` is that any
        previously sent record will have completed
        (e.g. Future.is_done() == True). A request is considered completed when
        either it is successfully acknowledged according to the 'acks'
        configuration for the producer, or it results in an error.

        Other threads can continue sending messages while one thread is blocked
        waiting for a flush call to complete; however, no guarantee is made
        about the completion of messages sent after the flush call begins.

        Arguments:
            timeout (float, optional): timeout in seconds to wait for completion.

        Raises:
            KafkaTimeoutError: failure to flush buffered records within the
                provided timeout

def flush(self, timeout=None):
        """
        Invoking this method makes all buffered records immediately available
        to send (even if linger_ms is greater than 0) and blocks on the
        completion of the requests associated with these records. The
        post-condition of :meth:`~kafka.KafkaProducer.flush` is that any
        previously sent record will have completed
        (e.g. Future.is_done() == True). A request is considered completed when
        either it is successfully acknowledged according to the 'acks'
        configuration for the producer, or it results in an error.

        Other threads can continue sending messages while one thread is blocked
        waiting for a flush call to complete; however, no guarantee is made
        about the completion of messages sent after the flush call begins.

        Arguments:
            timeout (float, optional): timeout in seconds to wait for completion.

        Raises:
            KafkaTimeoutError: failure to flush buffered records within the
                provided timeout
        """
        log.debug("Flushing accumulated records in producer.")  # trace
        self._accumulator.begin_flush()
        self._sender.wakeup()
        self._accumulator.await_flush_completion(timeout=timeout)