code: def flush(self, timeout=None):
          
          log.debug("Flushing accumulated records in producer.")  # trace
          self._accumulator.begin_flush()
          self._sender.wakeup()
          self._accumulator.await_flush_completion(timeout=timeout)
docs: 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