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

Code before:
            "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"
                    )
            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

Review comment (suggestion):
I think this whole logic can be simplified, why not check if the connection type is yarn OR spark, set the master with an inline if statement, otherwise, set the master connection accordingly

This can reduce nesting and make the logic simpler to understand and change in the future

Code after:
            "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:
                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")