async def METHOD_NAME(self, timeout_time=None):
    # flush all the buffered producer, the method will block until finishes or times out
    async with self._lock:
        futures = []
        for pid, producer in self._buffered_producers.items():
            # call each producer's flush method
            futures.append(
                (
                    pid,
                    asyncio.ensure_future(
                        producer.METHOD_NAME(timeout_time=timeout_time)
                    ),
                )
            )
        # gather results
        exc_results = {}
        for pid, future in futures:
            try:
                await future
            except Exception as exc:  # pylint: disable=broad-except
                exc_results[pid] = exc
        if not exc_results:
            _LOGGER.info("Flushing all partitions succeeded")
            return
        _LOGGER.warning(
            "Flushing all partitions partially failed with result %r.", exc_results
        )
        raise EventDataSendError(
            message="Flushing all partitions partially failed, failed partitions are {!r}"
            " Exception details are {!r}".format(exc_results.keys(), exc_results)
        )