docs: def _get_conn(self, host, port, afi):
          """Get or create a connection to a broker using host and port"""
          host_key = (host, port)
          if host_key not in self._conns:
              self._conns[host_key] = BrokerConnection(
                  host, port, afi,
                  request_timeout_ms=self.timeout * 1000,
                  client_id=self.client_id
              )
  
          conn = self._conns[host_key]
          if not conn.connect_blocking(self.timeout):
              conn.close()
              raise KafkaConnectionError("%s:%s (%s)" % (host, port, afi))
          return conn