def _send_broker_unaware_request(self, payloads, encoder_fn, decoder_fn):
        """
        Attempt to send a broker-agnostic request to one of the available
        brokers. Keep trying until you succeed.
        """
        hosts = set()
        for broker in self.brokers.values():
            host, port, afi = get_ip_port_afi(broker.host)
            hosts.add((host, broker.port, afi))

        hosts.update(self.hosts)
        hosts = list(hosts)
        random.shuffle(hosts)

        for (host, port, afi) in hosts:
            try:
                conn = self._get_conn(host, port, afi)
            except KafkaConnectionError:
                log.warning("Skipping unconnected connection: %s:%s (AFI %s)",
                            host, port, afi)
                continue
            request = encoder_fn(payloads=payloads)
            future = conn.send(request)

            # Block
            while not future.is_done:
                for r, f in conn.recv():
                    f.success(r)

            if future.failed():
                log.error("Request failed: %s", future.exception)
                continue

            return decoder_fn(future.value)

        raise KafkaUnavailableError('All servers failed to process request: %s' % (hosts,))