Repository: apache/airflow
File: providers/apache/spark/src/airflow/providers/apache/spark/hooks/spark_submit.py (Python)

Code before:
        initial spark-submit request.

        :return: if the driver status should be tracked
        """
        return "spark://" in self._connection["master"] and self._connection["deploy_mode"] == "cluster"

    def _resolve_connection(self) -> dict[str, Any]:
        # Build from connection master or default to yarn if not available
        conn_data = {
            "master": "yarn",
            "queue": None,  # yarn queue
            "deploy_mode": None,
            "spark_binary": self.spark_binary or DEFAULT_SPARK_BINARY,
            "namespace": None,
            "principal": self._principal,
            "keytab": self._keytab,
        }

        try:
            # Master can be local, yarn, spark://HOST:PORT, mesos://HOST:PORT and
            # k8s://https://<HOST>:<PORT>
            conn = self.get_connection(self._conn_id)

            # connection comes from UI or it's a spark-master
            if conn.conn_type == "spark":
                if conn.host and ("://" in conn.host or not conn.port):
                    conn_data["master"] = conn.host
                elif conn.port:  # spark master/standalone has port
                    conn_data["master"] = f"spark://{conn.host or ''}"
            else:
                if conn.conn_type != "yarn":
                    # For other conn_types (mesos, k8s, local, etc.): reconstruct URL
                    conn_data["master"] = f"{conn.conn_type}://{conn.host or ''}"
                else:
                    conn_data["master"] = conn.host

            # Append port if provided
            if conn.port and conn_data["master"]:
                conn_data["master"] += f":{conn.port}"

            # Determine optional yarn queue from the extra field
            extra = conn.extra_dejson
            conn_data["queue"] = self._yarn_queue if self._yarn_queue else extra.get("queue")
            conn_data["deploy_mode"] = self._deploy_mode if self._deploy_mode else extra.get("deploy-mode")
            if not self.spark_binary:
                self.spark_binary = extra.get("spark-binary", DEFAULT_SPARK_BINARY)
                if self.spark_binary is not None and self.spark_binary not in ALLOWED_SPARK_BINARIES:
                    raise ValueError(
                        f"The spark-binary extra can be one of {ALLOWED_SPARK_BINARIES} and it"
                        f" was `{self.spark_binary}`. Please make sure your spark binary is one of the"
                        f" allowed ones and that it is available on the PATH"

Review comment (bug):
Can there be a case where a host is set AND the port is not set? 

What if I use a custom secret backend and not the regular ui to store connections? I can see a problem here where the Conn type is spark, yet a port is not set and '://' is not in the host string, which causes nothing to happen and the application later fails on a key error

Code after:
        initial spark-submit request.

        :return: if the driver status should be tracked
        """
        return "spark://" in self._connection["master"] and self._connection["deploy_mode"] == "cluster"

    def _resolve_connection(self) -> dict[str, Any]:
        # Build from connection master or default to yarn if not available
        conn_data = {
            "master": "yarn",
            "queue": None,  # yarn queue
            "deploy_mode": None,
            "spark_binary": self.spark_binary or DEFAULT_SPARK_BINARY,
            "namespace": None,
            "principal": self._principal,
            "keytab": self._keytab,
        }

        try:
            # Master can be local, yarn, spark://HOST:PORT, mesos://HOST:PORT and
            # k8s://https://<HOST>:<PORT>
            conn = self.get_connection(self._conn_id)

            # When connection is created from URI, the scheme (spark://, k8s://, etc.)
            # is stored in conn_type, and conn.host contains only the hostname.
            # When created from UI, conn_type is typically "spark" and conn.host
            # may contain the full master URL (e.g., k8s://https://host).
            if conn.conn_type == "spark" and conn.host and ("://" in conn.host or not conn.port):
                # UI-based spark connection where host contains the full master URL
                conn_data["master"] = conn.host
            else:
                # Reconstruct URL with conn_type as protocol
                conn_data["master"] = f"{conn.conn_type}://{conn.host or ''}"

            # Append port if provided
            conn_data["master"] = f"{conn_data['master']}:{conn.port}" if conn.port else conn_data["master"]

            # Determine optional yarn queue from the extra field
            extra = conn.extra_dejson
            conn_data["queue"] = self._yarn_queue if self._yarn_queue else extra.get("queue")
            conn_data["deploy_mode"] = self._deploy_mode if self._deploy_mode else extra.get("deploy-mode")
            if not self.spark_binary:
                self.spark_binary = extra.get("spark-binary", DEFAULT_SPARK_BINARY)
                if self.spark_binary is not None and self.spark_binary not in ALLOWED_SPARK_BINARIES:
                    raise ValueError(
                        f"The spark-binary extra can be one of {ALLOWED_SPARK_BINARIES} and it"
                        f" was `{self.spark_binary}`. Please make sure your spark binary is one of the"
                        f" allowed ones and that it is available on the PATH"
                    )
            conn_spark_home = extra.get("spark-home")
            if conn_spark_home: