'Attempt to send a broker-agnostic request to one of the available
brokers. Keep trying until you succeed.'

def _send_broker_unaware_request(self, payloads, encoder_fn, decoder_fn):

    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 ConnectionError:
            log.warning('Skipping    unconnected    connection:    %s:%s    (AFI    %s)', host, port, afi)
            continue
        request = encoder_fn(payloads=payloads)
        future = conn.send(request)
        while (not future.is_done):
            conn.recv()
        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))
