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.
          """
          for (host, port) in self.hosts:
              requestId = self._next_id()
              log.debug('Request %s: %s', requestId, payloads)
              try:
                  conn = self._get_conn(host, port)
                  request = encoder_fn(client_id=self.client_id,
                                       correlation_id=requestId,
                                       payloads=payloads)
  
                  conn.send(requestId, request)
                  response = conn.recv(requestId)
                  decoded = decoder_fn(response)
                  log.debug('Response %s: %s', requestId, decoded)
                  return decoded
  
              except Exception:
                  log.exception('Error sending request [%s] to server %s:%s, '
                                'trying next server', requestId, host, port)
  
          raise KafkaUnavailableError('All servers failed to process request')