docs: 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,))