Oliver Ricken created AIRFLOW-7109:
--------------------------------------
Summary: HiveServer2Hook and pyHive do not handle timeout correctly
Key: AIRFLOW-7109
URL: https://issues.apache.org/jira/browse/AIRFLOW-7109
Project: Apache Airflow
Issue Type: Bug
Components: hooks, utils
Affects Versions: 1.10.6
Reporter: Oliver Ricken
Attachments: dag.py, dev_hive_hook.py, dev_hive_operator.py,
modified_HiveServer2Hook_task.log, taskinstance.py, timeout.py,
vanilla_HiverOperator_task.log
Dear experts,
using the 1.10.6 vanilla "HiveServer2Hook" from airflow/hooks/hive_hooks in a
custom-operator, we observe the improper handling of timeouts configured on
task-level using "execution_timeout".
The picture is always the same:
# The hook is instantiated, the connection opend and the query submitted
# The task-timeout is reached but the query is completed nonetheless
(confirmed by looking into YARN, job is flagged as "finished", i.e. successful)
# Although the query has been successful (hive-wise), no result is obtained
and processed but an obscure codec-error is raised, making the task fail. The
codec-error is based on the actual AirflowTimeoutException not being properly
propagated.
The persistent codec-error (including its stacktrace) is given below.
We investigated the issue further by slightly modifiying the vanilla version of
the HiveServer2Hook and adding some debugging output to various modules.
* We made the "pyhive.hive.connect"-object a class-variable (instead of having
it only exist in the scope of the \_get_results-classmethod (vanilla)
* We added a "kill()"-method that invokes "connection.close()" on said
class-variable and call the method upon execution of "on_kill()" in our
custom-operator.
* We added debugging (i.e. logging) output to the
airflow/models/taskinstance.py "_run_raw_task"-method and the
"airflow/utils/timeout.py" class's "__enter__()" and "__exit__()"-methods.
Our findings, based on the attached log-files (and the modified
core-module-files as well as the modified hook, the custom-operator and the
DAG) is the following:
* In order to invoke the "on_kill()"-method of any operator upon timeout of
the task,
the "timeout.py"-class used to execute the "task_copy" from ("taskinstance.py,
"_run_raw_task") needs to exit with an "AirflowTimeoutException" (and nothing
else!)
* The AirflowTimeoutException is produced but not at the point in runtime at
which the timeout has actually been reached but only at the end of the
pyHive-execution of the query.
* Upward propagation of the AirflowTimeoutException then fails with the
obscure codec-error mentioned earlier and thus no "on_kill()"- and
"kill()"-methods are invoked. Nota bena: this would not be necessary as the
task does not need actual killing anymore as it converged anyhow.
In out opinion this leads to two aspects to be studied, understood and fixed,
potentially in pyHive though.
* The execution of the pyHive query needs to be killed upon timeout
* The propagation of the exception needs be performed properly
NB: closing the pyhive.hive.connection-object upon invocation of "on_kill()"
does work properly if the task is killed manually, e.g. by setting it to
"failed". In this case, "on_kill()" is invoked directly instead of waiting for
the AirflowTimeoutException to pop up.
NB: things work perfectly using the "HiverCLIHook" which, under the hood, uses
beeline executed from a python subprocess-call which is terminated (killed
after 60'') upon invocation of "on_kill" in e.g. the vanilla HiveOperator. The
log attached as "vanilla_HiverOperator_task.log" shows how the behaviour of the
vanilla HiverOpator performing the same query. In principle, similarly flawless
behaviour is expected when using the HiveServer2Hook to submit the query ...
Your input on this matter and any suggestions are highly appreciated. The
"HiveServer2Hook" is, in our opinion, an irreplacable tool when in need of
query-results for postprocessing.
Cheers
{{[2020-03-20 14:45:42,982] \{taskinstance.py:1060} ERROR - 'utf-8' codec can't
decode byte 0xdb in position 0: invalid continuation byte
Traceback (most recent call last):
File "/home/airflow/dags/src/vipa_import/plugins/hooks/dev_hive_hook.py",
line 33, in _get_results
cur.execute(statement)
File "/home/airflow/airflow_venv/lib/python3.6/site-packages/pyhive/hive.py",
line 364, in execute
response = self._connection.client.ExecuteStatement(req)
File
"/home/airflow/airflow_venv/lib/python3.6/site-packages/TCLIService/TCLIService.py",
line 280, in ExecuteStatement
return self.recv_ExecuteStatement()
File
"/home/airflow/airflow_venv/lib/python3.6/site-packages/TCLIService/TCLIService.py",
line 292, in recv_ExecuteStatement
(fname, mtype, rseqid) = iprot.readMessageBegin()
File
"/home/airflow/airflow_venv/lib/python3.6/site-packages/thrift/protocol/TBinaryProtocol.py",
line 134, in readMessageBegin
sz = self.readI32()
File
"/home/airflow/airflow_venv/lib/python3.6/site-packages/thrift/protocol/TBinaryProtocol.py",
line 217, in readI32
buff = self.trans.readAll(4)
File
"/home/airflow/airflow_venv/lib/python3.6/site-packages/thrift/transport/TTransport.py",
line 60, in readAll
chunk = self.read(sz - have)
File
"/home/airflow/airflow_venv/lib/python3.6/site-packages/thrift_sasl/__init__.py",
line 166, in read
self._read_frame()
File
"/home/airflow/airflow_venv/lib/python3.6/site-packages/thrift_sasl/__init__.py",
line 170, in _read_frame
header = self._trans.readAll(4)
File
"/home/airflow/airflow_venv/lib/python3.6/site-packages/thrift/transport/TTransport.py",
line 60, in readAll
chunk = self.read(sz - have)
File
"/home/airflow/airflow_venv/lib/python3.6/site-packages/thrift/transport/TSocket.py",
line 117, in read
buff = self.handle.recv(sz)
File
"/home/airflow/airflow_venv/lib/python3.6/site-packages/airflow/utils/timeout.py",
line 43, in handle_timeout
raise AirflowTaskTimeout(self.error_message)
airflow.exceptions.AirflowTaskTimeout: Timeout, PID: 17089}}
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
