def METHOD_NAME(self, *, flush=True, timeout_time=None, raise_error=False):
    with self._lock:
        futures = []
        # stop all buffered producers
        for pid, producer in self._buffered_producers.items():
            futures.append(
                (
                    pid,
                    self._executor.submit(
                        producer.stop,
                        flush=flush,
                        timeout_time=timeout_time,
                        raise_error=raise_error,
                    ),
                )
            )
        exc_results = {}
        # gather results
        for pid, future in futures:
            try:
                future.result()
            except Exception as exc:  # pylint: disable=broad-except
                exc_results[pid] = exc
        if exc_results:
            _LOGGER.warning(
                "Stopping all partitions partially failed with result %r.",
                exc_results,
            )
            if raise_error:
                raise EventHubError(
                    message="Stopping all partitions partially failed, failed partitions are {!r}"
                    " Exception details are {!r}".format(
                        exc_results.keys(), exc_results
                    )
                )
        if not self._existing_executor:
            self._executor.shutdown()