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')