docs: def await_flush_completion(self, timeout=None):
          """
          Mark all partitions as ready to send and block until the send is complete
          """
          try:
              for batch in self._incomplete.all():
                  log.debug('Waiting on produce to %s',
                            batch.produce_future.topic_partition)
                  if not batch.produce_future.wait(timeout=timeout):
                      raise Errors.KafkaTimeoutError('Timeout waiting for future')
                  if not batch.produce_future.is_done:
                      raise Errors.UnknownError('Future not done')
  
                  if batch.produce_future.failed():
                      log.warning(batch.produce_future.exception)
          finally:
              self._flushes_in_progress.decrement()