'Mark all partitions as ready to send and block until the send is complete'

def await_flush_completion(self, timeout=None):

    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()
