spark-submit: replace principle by principal (#44150)

* spark-submit: replace `principle` by `principal`

This is related to #43679, in which the `airflow.security.kerberos.get_kerberos_principle` function was renamed `get_kerberos_principle`.

In this patch, we introduce a `try/except` block around this import,
getting ready to deprecate the typo-ed function in Airflow 3.0.

We also fix a innocuous typo in a unit test.

Signed-off-by: Balthazar Rouberol <brouberol@wikimedia.org>

* Update providers/src/airflow/providers/apache/spark/hooks/spark_submit.py

* Update providers/src/airflow/providers/apache/spark/hooks/spark_submit.py

* Update providers/src/airflow/providers/apache/spark/hooks/spark_submit.py

* Update providers/src/airflow/providers/apache/spark/hooks/spark_submit.py

* Update providers/src/airflow/providers/apache/spark/hooks/spark_submit.py

---------

Signed-off-by: Balthazar Rouberol <brouberol@wikimedia.org>
Co-authored-by: Elad Kalif <45845474+eladkal@users.noreply.github.com>
Co-authored-by: rom sharon <33751805+romsharon98@users.noreply.github.com>

providers/src/airflow/providers/apache/spark/hooks/spark_submit.py

@@ -518,9 +518,15 @@ def _build_track_driver_status_command(self) -> list[str]:
 
     def _resolve_kerberos_principal(self, principal: str | None) -> str:
         """Resolve kerberos principal."""
-        from airflow.security.kerberos import get_kerberos_principle
+        # todo: remove try/exception when min airflow version is 3.0
+        try:
+            from airflow.security.kerberos import get_kerberos_principal  # type: ignore[attr-defined]
+        except ImportError:
+            from airflow.security.kerberos import (
+                get_kerberos_principle as get_kerberos_principal,  # type: ignore[attr-defined]
+            )
 
-        return get_kerberos_principle(principal)
+        return get_kerberos_principal(principal)
 
     def submit(self, application: str = "", **kwargs: Any) -> None:
         """

providers/tests/apache/spark/hooks/test_spark_submit.py

@@ -230,11 +230,11 @@ def test_build_spark_submit_command(self, mock_get_env):
 
     @patch("airflow.configuration.conf.get_mandatory_value")
     def test_resolve_spark_submit_env_vars_use_krb5ccache_missing_principal(self, mock_get_madantory_value):
-        mock_principle = "airflow"
-        mock_get_madantory_value.return_value = mock_principle
+        mock_principal = "airflow"
+        mock_get_madantory_value.return_value = mock_principal
         hook = SparkSubmitHook(conn_id="spark_yarn_cluster", principal=None, use_krb5ccache=True)
         mock_get_madantory_value.assert_called_with("kerberos", "principal")
-        assert hook._principal == mock_principle
+        assert hook._principal == mock_principal
 
     def test_resolve_spark_submit_env_vars_use_krb5ccache_missing_KRB5CCNAME_env(self):
         hook = SparkSubmitHook(