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
sasl_client = sasl.Client()
                sasl_client.setAttr("host", conn.host)
                sasl_client.setAttr("service", kerberos_service_name)
                sasl_client.init()
                return sasl_client
            from thrift_sasl import TSaslClientTransport
            transport = TSaslClientTransport(sasl_factory, "GSSAPI", conn_socket)
        else:
            transport = TTransport.TBufferedTransport(conn_socket)
        protocol = TBinaryProtocol.TBinaryProtocol(transport)
        return hmsclient.HMSClient(iprot=protocol)
    def _find_valid_server(self) -> Any:
        conns = self.get_connections(self.conn_id)
        for conn in conns:
            host_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            self.log.info("Trying to connect to %s:%s", conn.host, conn.port)
            if host_socket.connect_ex((conn.host, conn.port)) == 0:
                self.log.info("Connected to %s:%s", conn.host, conn.port)
                host_socket.close()
                return conn
            else:
                self.log.error("Could not connect to %s:%s", conn.host, conn.port)
        return None
    def get_conn(self) -> Any: