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

Code before:
            "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"
                    )
            conn_spark_home = extra.get("spark-home")
            if conn_spark_home:
                raise ValueError(
                    "The `spark-home` extra is not allowed any more. Please make sure one of"
                    f" {ALLOWED_SPARK_BINARIES} is available on the PATH, and set `spark-binary`"
                    " if needed."
                )
            conn_data["spark_binary"] = self.spark_binary
            conn_data["namespace"] = extra.get("namespace")
            if conn_data["principal"] is None:
                conn_data["principal"] = extra.get("principal")

Review comment (suggestion):
The `conn_data` check will result in a keyerror if  the master was not set, better use get here, and this can also be done inline

Code after:
            "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:
                raise ValueError(
                    "The `spark-home` extra is not allowed any more. Please make sure one of"
                    f" {ALLOWED_SPARK_BINARIES} is available on the PATH, and set `spark-binary`"
                    " if needed."
                )
            conn_data["spark_binary"] = self.spark_binary
            conn_data["namespace"] = extra.get("namespace")
            if conn_data["principal"] is None:
                conn_data["principal"] = extra.get("principal")
            if conn_data["keytab"] is None:
                base64_keytab = extra.get("keytab")
                if base64_keytab is not None: