def _send_broker_unaware_request(self, payloads, encoder_fn, decoder_fn):
for (host, port) in self.hosts:<EOL><INDENT>requestId = self._next_id()<EOL>log.debug('<STR_LIT>', requestId, payloads)<EOL>try:<EOL><INDENT>conn = self._get_conn(host, port)<EOL>request = encoder_fn(client_id=self.client_id,<EOL>correlation_id=requestId,<EOL>payloads=payloads)<EOL>conn.send(requestId, request)<EOL>response = conn.recv(requestId)<EOL>decoded = decoder_fn(response)<EOL>log.debug('<STR_LIT>', requestId, decoded)<EOL>return decoded<EOL><DEDENT>except Exception:<EOL><INDENT>log.exception('<STR_LIT>'<EOL>'<STR_LIT>', requestId, host, port)<EOL><DEDENT><DEDENT>raise KafkaUnavailableError('<STR_LIT>')<EOL>
Attempt to send a broker-agnostic request to one of the available
brokers. Keep trying until you succeed.