[jira] [Commented] (AIRFLOW-2511) Subdag failed by scheduler deadlock

["Trevor Edwards (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/AIRFLOW-2511?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16599024#comment-16599024 ] 

Trevor Edwards commented on AIRFLOW-2511:
-----------------------------------------

Doing some digging into this issue, it seems like this kind of error should only happen if sessions are being misused: http://docs.sqlalchemy.org/en/latest/faq/sessions.html#this-session-s-transaction-has-been-rolled-back-due-to-a-previous-exception-during-flush-or-similar

 

Reviewing 1.10 (though 1.9.0 is similar) https://github.com/apache/incubator-airflow/blob/1.10.0/airflow/models.py#L1602-L1673, it seems like there are several calls to methods which call session.commit() but do not catch exceptions and rollback. The broad except at the end of the highlighted region in my link also attempts to do a session.commit() in the method call, meaning if we swallow a sql exception in order to get to that line, we'll get this exception. I'm not convinced this is what caused this specific issue, but it is at least related and probably an issue as well. I'll keep digging.

> Subdag failed by scheduler deadlock
> -----------------------------------
>
>                 Key: AIRFLOW-2511
>                 URL: https://issues.apache.org/jira/browse/AIRFLOW-2511
>             Project: Apache Airflow
>          Issue Type: Bug
>    Affects Versions: 1.9.0
>            Reporter: Yohei Onishi
>            Priority: Major
>
> I am using subdag and sometimes main dag marked failed because of the following error. In this case, tasks in the subdag stopped.
> {code:java}
> hourly_dag = DAG(
>   hourly_dag_name,
>   default_args=dag_default_args,
>   params=dag_custom_params,
>   schedule_interval=config_values.hourly_job_interval,
>   max_active_runs=2)
> hourly_subdag = SubDagOperator(
>   task_id='s3_to_hive',
>   subdag=LoadFromS3ToHive(
>   hourly_dag,
>   's3_to_hive'),
>   dag=hourly_dag)
> {code}
> I got this error in main dag. bug in scheduler?
> {code:java}
> [2018-05-22 21:52:19,683] {models.py:1595} ERROR - This Session's transaction has been rolled back due to a previous exception during flush. To begin a new transaction with this Session, first issue Session.rollback(). Original exception was: (_mysql_exceptions.OperationalError) (1213, 'Deadlock found when trying to get lock; try restarting transaction') [SQL: 'UPDATE task_instance SET state=%s WHERE task_instance.task_id = %s AND task_instance.dag_id = %s AND task_instance.execution_date = %s'] [parameters: ('queued', 'transfer_from_tmp_table_into_cleaned_table', 'rfid_warehouse_carton_wh_g_dl_dwh_csv_uqjp_1h.s3_to_hive', datetime.datetime(2018, 5, 7, 5, 2))] (Background on this error at: http://sqlalche.me/e/e3q8)
> Traceback (most recent call last):
> sqlalchemy.exc.InvalidRequestError: This Session's transaction has been rolled back due to a previous exception during flush. To begin a new transaction with this Session, first issue Session.rollback(). Original exception was: (_mysql_exceptions.OperationalError) (1213, 'Deadlock found when trying to get lock; try restarting transaction') [SQL: 'UPDATE task_instance SET state=%s WHERE task_instance.task_id = %s AND task_instance.dag_id = %s AND task_instance.execution_date = %s'] [parameters: ('queued', 'transfer_from_tmp_table_into_cleaned_table', 'rfid_warehouse_carton_wh_g_dl_dwh_csv_uqjp_1h.s3_to_hive', datetime.datetime(2018, 5, 7, 5, 2))] (Background on this error at: http://sqlalche.me/e/e3q8)
> [2018-05-22 21:52:19,687] {models.py:1624} INFO - Marking task as FAILED.
> [2018-05-22 21:52:19,688] {base_task_runner.py:98} INFO - Subtask: [2018-05-22 21:52:19,688] {slack_hook.py:143} INFO - Message is prepared: 
> [2018-05-22 21:52:19,688] {base_task_runner.py:98} INFO - Subtask: {"attachments": [{"color": "danger", "text": "", "fields": [{"title": "DAG", "value": "<https://data-platform-airflow-tokyo.data.fastretailing.com:8080/admin/airflow/tree?dag_id=rfid_warehouse_carton_wh_g_dl_dwh_csv_uqjp_1h|rfid_warehouse_carton_wh_g_dl_dwh_csv_uqjp_1h>", "short": true}, {"title": "Owner", "value": "airflow", "short": true}, {"title": "Task", "value": "s3_to_hive", "short": false}, {"title": "Status", "value": "FAILED", "short": false}, {"title": "Execution Time", "value": "2018-05-07T05:02:00", "short": true}, {"title": "Duration", "value": "826.305929", "short": true}, {"value": "<https://data-platform-airflow-tokyo.data.fastretailing.com:8080/admin/airflow/log?dag_id=rfid_warehouse_carton_wh_g_dl_dwh_csv_uqjp_1h&task_id=s3_to_hive&execution_date=2018-05-07T05:02:00|View Task Log>", "short": false}]}]}
> [2018-05-22 21:52:19,688] {models.py:1638} ERROR - Failed at executing callback
> [2018-05-22 21:52:19,688] {models.py:1639} ERROR - This Session's transaction has been rolled back due to a previous exception during flush. To begin a new transaction with this Session, first issue Session.rollback(). Original exception was: (_mysql_exceptions.OperationalError) (1213, 'Deadlock found when trying to get lock; try restarting transaction') [SQL: 'UPDATE task_instance SET state=%s WHERE task_instance.task_id = %s AND task_instance.dag_id = %s AND task_instance.execution_date = %s'] [parameters: ('queued', 'transfer_from_tmp_table_into_cleaned_table', 'rfid_warehouse_carton_wh_g_dl_dwh_csv_uqjp_1h.s3_to_hive', datetime.datetime(2018, 5, 7, 5, 2))] (Background on this error at: http://sqlalche.me/e/e3q8)
> {code}
>  



--
This message was sent by Atlassian JIRA
(v7.6.3#76005)