https://github.com/apache/airflow/blob/6c458f29c0eeadb1282e524e76fdd379d6436824/airflow/providers/apache/hive/hooks/hive.py#L464

It should be:
```
if hasattr(self, 'sub_process')
```
airflow/providers/apache/hive/hooks/hive.py
self,
        hive_cli_conn_id: str = default_conn_name,
        run_as: Optional[str] = None,
        mapred_queue: Optional[str] = None,
        mapred_queue_priority: Optional[str] = None,
        mapred_job_name: Optional[str] = None,
    ) -> None:
        super().__init__()
        conn = self.get_connection(hive_cli_conn_id)
        self.hive_cli_params: str = conn.extra_dejson.get('hive_cli_params', '')
        self.use_beeline: bool = conn.extra_dejson.get('use_beeline', False)
        self.auth = conn.extra_dejson.get('auth', 'noSasl')
        self.conn = conn
        self.run_as = run_as
        self.sub_process: Any = None
        if mapred_queue_priority:
            mapred_queue_priority = mapred_queue_priority.upper()
            if mapred_queue_priority not in HIVE_QUEUE_PRIORITIES:
                raise AirflowException(
                    "Invalid Mapred Queue Priority.  Valid values are: "
                    "{}".format(', '.join(HIVE_QUEUE_PRIORITIES))
                )
        self.mapred_queue = mapred_queue or conf.get('hive', 'default_hive_mapred_queue')
        self.mapred_queue_priority = mapred_queue_priority
        self.mapred_job_name = mapred_job_name
    def _get_proxy_user(self) -> str: