Fix HiveServer2Hook password handling for PLAIN auth (#62338)

providers/apache/hive/src/airflow/providers/apache/hive/hooks/hive.py

@@ -877,8 +877,8 @@ def get_conn(self, schema: str | None = None) -> Any:
             auth_mechanism = db.extra_dejson.get("auth_mechanism", "KERBEROS")
             kerberos_service_name = db.extra_dejson.get("kerberos_service_name", "hive")
 
-        # Password should be set if and only if in LDAP or CUSTOM mode
-        if auth_mechanism in ("LDAP", "CUSTOM"):
+        # Password should be set if in LDAP, CUSTOM or PLAIN mode
+        if auth_mechanism in ("LDAP", "CUSTOM", "PLAIN"):
             password = db.password
 
         from pyhive.hive import connect

providers/apache/hive/tests/unit/apache/hive/hooks/test_hive.py

@@ -677,6 +677,26 @@ def test_get_conn_with_password(self, mock_connect):
                 database="default",
             )
 
+    @mock.patch("pyhive.hive.connect")
+    def test_get_conn_with_password_plain(self, mock_connect):
+        conn_id = "conn_plain_with_password"
+        conn_env = CONN_ENV_PREFIX + conn_id.upper()
+
+        with mock.patch.dict(
+            "os.environ",
+            {conn_env: "jdbc+hive2://login:password@localhost:10000/default?auth_mechanism=PLAIN"},
+        ):
+            HiveServer2Hook(hiveserver2_conn_id=conn_id).get_conn()
+            mock_connect.assert_called_once_with(
+                host="localhost",
+                port=10000,
+                auth="PLAIN",
+                kerberos_service_name=None,
+                username="login",
+                password="password",
+                database="default",
+            )
+
     @pytest.mark.parametrize(
         ("host", "port", "schema", "message"),
         [