def _send_bootstrap_request(self, request):
        """Make a request using an ephemeral broker connection

        This routine is used to make broker-unaware requests to get the initial
        cluster metadata. It cycles through the configured hosts, trying to
        connect and send the request to each in turn. This temporary connection
        is closed once a response is received.

        Note that most Kafka APIs require requests be sent to a specific
        broker. This method will only function for broker-agnostic requests
        like:

          * `Metadata <https://kafka.apache.org/protocol.html#The_Messages_Metadata>`_
          * `FindCoordinator <https://kafka.apache.org/protocol.html#The_Messages_FindCoordinator>`_

        :param bytes request:
            The bytes of a Kafka `RequestMessage`_ structure. It must have
            a unique (to this connection) correlation ID.

        :returns: API response message for *request*
        :rtype: Deferred[bytes]

        :raises:
            - `KafkaUnavailableError` when making the request of all known hosts
               has failed.
            - `twisted.internet.defer.TimeoutError` when connecting or making
               a request exceeds the timeout.
        """
        hostports = list(self._bootstrap_hosts)
        random.shuffle(hostports)
        for host, port in hostports:
            ep = self._endpoint_factory(self.reactor, host, port)
            try:
                protocol = yield ep.connect(_bootstrapFactory)
            except Exception as e:
                log.debug("%s: bootstrap connect to %s:%s -> %s", self, host, port, e)
                continue

            try:
                response = yield protocol.request(request).addTimeout(self.timeout, self.reactor)
            except Exception:
                log.debug("%s: bootstrap request to %s:%s failed", self, host, port, exc_info=True)
            else:
                returnValue(response)
            finally:
                protocol.transport.loseConnection()

        raise KafkaUnavailableError("Failed to bootstrap from hosts {}".format(hostports))
Make a request using an ephemeral broker connection

        This routine is used to make broker-unaware requests to get the initial
        cluster metadata. It cycles through the configured hosts, trying to
        connect and send the request to each in turn. This temporary connection
        is closed once a response is received.

        Note that most Kafka APIs require requests be sent to a specific
        broker. This method will only function for broker-agnostic requests
        like:

          * `Metadata <https://kafka.apache.org/protocol.html#The_Messages_Metadata>`_
          * `FindCoordinator <https://kafka.apache.org/protocol.html#The_Messages_FindCoordinator>`_

        :param bytes request:
            The bytes of a Kafka `RequestMessage`_ structure. It must have
            a unique (to this connection) correlation ID.

        :returns: API response message for *request*
        :rtype: Deferred[bytes]

        :raises:
            - `KafkaUnavailableError` when making the request of all known hosts
               has failed.
            - `twisted.internet.defer.TimeoutError` when connecting or making
               a request exceeds the timeout.