Get or create a connection to a broker using host and port

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