You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "ASF GitHub Bot (Jira)" <ji...@apache.org> on 2019/12/01 19:41:00 UTC
[jira] [Commented] (AIRFLOW-6141) The sensor cannot be executed by
invalid deps
[ https://issues.apache.org/jira/browse/AIRFLOW-6141?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16985660#comment-16985660 ]
ASF GitHub Bot commented on AIRFLOW-6141:
-----------------------------------------
mik-laj commented on pull request #6704: [AIRFLOW-6141] Remove ReadyToRescheduleDep if sensor mode == poke
URL: https://github.com/apache/airflow/pull/6704
I am trying to fix [AIRFLOW-6141](https://issues.apache.org/jira/browse/AIRFLOW-6141)I saw that this rule is enabled when the sensor mode is not correct. It now works correctly in both modes.
---
Make sure you have checked _all_ steps below.
### Jira
- [ ] My PR addresses the following [Airflow Jira](https://issues.apache.org/jira/browse/AIRFLOW/) issues and references them in the PR title. For example, "\[AIRFLOW-XXX\] My Airflow PR"
- https://issues.apache.org/jira/browse/AIRFLOW-6141
- In case you are fixing a typo in the documentation you can prepend your commit with \[AIRFLOW-XXX\], code changes always need a Jira issue.
- In case you are proposing a fundamental code change, you need to create an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)).
- In case you are adding a dependency, check if the license complies with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
### Description
- [ ] Here are some details about my PR, including screenshots of any UI changes:
### Tests
- [ ] My PR adds the following unit tests __OR__ does not need testing for this extremely good reason:
### Commits
- [ ] My commits all reference Jira issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)":
1. Subject is separated from body by a blank line
1. Subject is limited to 50 characters (not including Jira issue reference)
1. Subject does not end with a period
1. Subject uses the imperative mood ("add", not "adding")
1. Body wraps at 72 characters
1. Body explains "what" and "why", not "how"
### Documentation
- [ ] In case of new functionality, my PR adds documentation that describes how to use it.
- All the public functions and the classes in the PR contain docstrings that explain what it does
- If you implement backwards incompatible changes, please leave a note in the [Updating.md](https://github.com/apache/airflow/blob/master/UPDATING.md) so we can assign it to a appropriate release
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
> The sensor cannot be executed by invalid deps
> ---------------------------------------------
>
> Key: AIRFLOW-6141
> URL: https://issues.apache.org/jira/browse/AIRFLOW-6141
> Project: Apache Airflow
> Issue Type: Bug
> Components: backfill
> Affects Versions: 1.10.6
> Reporter: Kamil Bregula
> Priority: Major
>
> Hello,
> I have following DAG:
> {code:python}
> """Example DAG demonstrating the usage of the BranchPythonOperator."""
> import random
> from os import path
> import airflow
> from airflow.contrib.sensors.python_sensor import PythonSensor
> from airflow.models import DAG
> from airflow.operators.dummy_operator import DummyOperator
> from airflow.operators.python_operator import BranchPythonOperator
> dag = DAG(
> dag_id='demo_deadlock_backfill_issues',
> default_args={
> 'owner': 'Airflow',
> 'start_date': airflow.utils.dates.days_ago(2),
> },
> schedule_interval="@daily",
> )
> DEADLOCK_SHARED_SHARE = "/tmp/deadlock-shared-state"
> def python_callable():
> return False
> run = PythonSensor(
> task_id='run',
> dag=dag,
> python_callable=python_callable,
> ){code}
>
> I executed following command
> {code:bash}
> AIRFLOW__CORE__LOGGING_LEVEL=debug airflow dags backfill demo_deadlock_backfill_issues -s 2019-01-02 -e 2019-01-02 --reset_dagruns
> {code}
> I then got the following log:
> {code:java}
> /opt/airflow/airflow/models/dagbag.py:21: DeprecationWarning: the imp module is deprecated in favour of importlib; see the module's documentation for alternative uses import imp [2019-12-01 19:03:40,197] {settings.py:139} DEBUG - Setting up DB connection pool (PID 2429) [2019-12-01 19:03:40,198] {settings.py:180} INFO - settings.configure_orm(): Using pool settings. pool_size=5, max_overflow=10, pool_recycle=1800, pid=2429 [2019-12-01 19:03:40,269] {cli_action_loggers.py:41} DEBUG - Adding <function default_action_log at 0x7fb41b4eff28> to pre execution callback [2019-12-01 19:03:40,344] {cli_action_loggers.py:67} DEBUG - Calling callbacks: [<function default_action_log at 0x7fb41b4eff28>] [2019-12-01 19:03:40,361] {__init__.py:55} INFO - Using executor SequentialExecutor [2019-12-01 19:03:40,362] {dagbag.py:401} INFO - Filling up the DagBag from /root/airflow/dags [..] You are about to delete these 1 tasks: <TaskInstance: demo_deadlock_backfill_issues.run 2019-01-02 00:00:00+00:00 [running]>Are you sure? (yes/no): yes [2019-12-01 19:03:45,075] {backfill_job.py:396} DEBUG - *** Clearing out not_ready list *** [2019-12-01 19:03:45,085] {backfill_job.py:414} DEBUG - Task instance to run <TaskInstance: demo_deadlock_backfill_issues.run 2019-01-02 00:00:00+00:00 [shutdown]> state shutdown [2019-12-01 19:03:45,091] {taskinstance.py:631} DEBUG - <TaskInstance: demo_deadlock_backfill_issues.run 2019-01-02 00:00:00+00:00 [shutdown]> dependency 'Task Instance Not Running' PASSED: True, Task is not in running state. [2019-12-01 19:03:45,091] {taskinstance.py:631} DEBUG - <TaskInstance: demo_deadlock_backfill_issues.run 2019-01-02 00:00:00+00:00 [shutdown]> dependency 'Trigger Rule' PASSED: True, The task instance did not have any upstream tasks. [2019-12-01 19:03:45,091] {taskinstance.py:631} DEBUG - <TaskInstance: demo_deadlock_backfill_issues.run 2019-01-02 00:00:00+00:00 [shutdown]> dependency 'Previous Dagrun State' PASSED: True, The task did not have depends_on_past set. [2019-12-01 19:03:45,092] {taskinstance.py:631} DEBUG - <TaskInstance: demo_deadlock_backfill_issues.run 2019-01-02 00:00:00+00:00 [shutdown]> dependency 'Ready To Reschedule' PASSED: True, The task instance is not in State_UP_FOR_RESCHEDULE or NONE state. [2019-12-01 19:03:45,092] {taskinstance.py:631} DEBUG - <TaskInstance: demo_deadlock_backfill_issues.run 2019-01-02 00:00:00+00:00 [shutdown]> dependency 'Not In Retry Period' PASSED: True, The task instance was not marked for retrying. [2019-12-01 19:03:45,095] {taskinstance.py:631} DEBUG - <TaskInstance: demo_deadlock_backfill_issues.run 2019-01-02 00:00:00+00:00 [shutdown]> dependency 'Task Instance State' PASSED: False, Task is in the 'shutdown' state which is not a valid state for execution. The task must be cleared in order to be run. [2019-12-01 19:03:45,095] {taskinstance.py:608} DEBUG - Dependencies not met for <TaskInstance: demo_deadlock_backfill_issues.run 2019-01-02 00:00:00+00:00 [shutdown]>, dependency 'Task Instance State' FAILED: Task is in the 'shutdown' state which is not a valid state for execution. The task must be cleared in order to be run. [2019-12-01 19:03:45,095] {backfill_job.py:536} DEBUG - Adding <TaskInstance: demo_deadlock_backfill_issues.run 2019-01-02 00:00:00+00:00 [shutdown]> to not_ready [2019-12-01 19:03:50,064] {base_job.py:196} DEBUG - [heartbeat] [2019-12-01 19:03:50,065] {base_executor.py:121} DEBUG - 0 running task instances [2019-12-01 19:03:50,070] {base_executor.py:122} DEBUG - 0 in queue [2019-12-01 19:03:50,071] {base_executor.py:123} DEBUG - 32 open slots [2019-12-01 19:03:50,072] {base_executor.py:132} DEBUG - Calling the <class 'airflow.executors.sequential_executor.SequentialExecutor'> sync method [2019-12-01 19:03:50,075] {backfill_job.py:598} WARNING - Deadlock discovered for ti_status.to_run=odict_values([<TaskInstance: demo_deadlock_backfill_issues.run 2019-01-02 00:00:00+00:00 [shutdown]>]) [2019-12-01 19:03:50,081] {dagrun.py:272} DEBUG - Updating state for <DagRun demo_deadlock_backfill_issues @ 2019-01-02 00:00:00+00:00: backfill_2019-01-02T00:00:00+00:00, externally triggered: False> considering 1 task(s) [2019-12-01 19:03:50,085] {taskinstance.py:631} DEBUG - <TaskInstance: demo_deadlock_backfill_issues.run 2019-01-02 00:00:00+00:00 [shutdown]> dependency 'Trigger Rule' PASSED: True, The task instance did not have any upstream tasks. [2019-12-01 19:03:50,086] {taskinstance.py:631} DEBUG - <TaskInstance: demo_deadlock_backfill_issues.run 2019-01-02 00:00:00+00:00 [shutdown]> dependency 'Previous Dagrun State' PASSED: True, The task did not have depends_on_past set. [2019-12-01 19:03:50,086] {taskinstance.py:631} DEBUG - <TaskInstance: demo_deadlock_backfill_issues.run 2019-01-02 00:00:00+00:00 [shutdown]> dependency 'Ready To Reschedule' PASSED: True, The context specified that being in a reschedule period was permitted. [2019-12-01 19:03:50,086] {taskinstance.py:631} DEBUG - <TaskInstance: demo_deadlock_backfill_issues.run 2019-01-02 00:00:00+00:00 [shutdown]> dependency 'Not In Retry Period' PASSED: True, The context specified that being in a retry period was permitted. [2019-12-01 19:03:50,086] {taskinstance.py:614} DEBUG - Dependencies all met for <TaskInstance: demo_deadlock_backfill_issues.run 2019-01-02 00:00:00+00:00 [shutdown]> [2019-12-01 19:03:50,090] {backfill_job.py:359} INFO - [backfill progress] | finished run 0 of 1 | tasks waiting: 0 | succeeded: 0 | running: 0 | failed: 0 | skipped: 0 | deadlocked: 1 | not ready: 1 [2019-12-01 19:03:50,090] {backfill_job.py:364} DEBUG - Finished dag run loop iteration. Remaining tasks odict_values([]) [2019-12-01 19:03:50,091] {taskinstance.py:631} DEBUG - <TaskInstance: demo_deadlock_backfill_issues.run 2019-01-02 00:00:00+00:00 [shutdown]> dependency 'Trigger Rule' PASSED: True, The task instance did not have any upstream tasks. [2019-12-01 19:03:50,091] {taskinstance.py:631} DEBUG - <TaskInstance: demo_deadlock_backfill_issues.run 2019-01-02 00:00:00+00:00 [shutdown]> dependency 'Previous Dagrun State' PASSED: True, The task did not have depends_on_past set. [2019-12-01 19:03:50,091] {taskinstance.py:631} DEBUG - <TaskInstance: demo_deadlock_backfill_issues.run 2019-01-02 00:00:00+00:00 [shutdown]> dependency 'Ready To Reschedule' PASSED: True, The task instance is not in State_UP_FOR_RESCHEDULE or NONE state. [2019-12-01 19:03:50,091] {taskinstance.py:631} DEBUG - <TaskInstance: demo_deadlock_backfill_issues.run 2019-01-02 00:00:00+00:00 [shutdown]> dependency 'Not In Retry Period' PASSED: True, The task instance was not marked for retrying. [2019-12-01 19:03:50,092] {taskinstance.py:614} DEBUG - Dependencies all met for <TaskInstance: demo_deadlock_backfill_issues.run 2019-01-02 00:00:00+00:00 [shutdown]> [2019-12-01 19:03:50,092] {taskinstance.py:631} DEBUG - <TaskInstance: demo_deadlock_backfill_issues.run 2019-01-02 00:00:00+00:00 [shutdown]> dependency 'Trigger Rule' PASSED: True, The task instance did not have any upstream tasks. [2019-12-01 19:03:50,093] {taskinstance.py:631} DEBUG - <TaskInstance: demo_deadlock_backfill_issues.run 2019-01-02 00:00:00+00:00 [shutdown]> dependency 'Previous Dagrun State' PASSED: True, The context specified that the state of past DAGs could be ignored. [2019-12-01 19:03:50,093] {taskinstance.py:631} DEBUG - <TaskInstance: demo_deadlock_backfill_issues.run 2019-01-02 00:00:00+00:00 [shutdown]> dependency 'Ready To Reschedule' PASSED: True, The task instance is not in State_UP_FOR_RESCHEDULE or NONE state. [2019-12-01 19:03:50,094] {taskinstance.py:631} DEBUG - <TaskInstance: demo_deadlock_backfill_issues.run 2019-01-02 00:00:00+00:00 [shutdown]> dependency 'Not In Retry Period' PASSED: True, The task instance was not marked for retrying. [2019-12-01 19:03:50,095] {taskinstance.py:614} DEBUG - Dependencies all met for <TaskInstance: demo_deadlock_backfill_issues.run 2019-01-02 00:00:00+00:00 [shutdown]> [2019-12-01 19:03:50,097] {base_executor.py:121} DEBUG - 0 running task instances [2019-12-01 19:03:50,097] {base_executor.py:122} DEBUG - 0 in queue [2019-12-01 19:03:50,098] {base_executor.py:123} DEBUG - 32 open slots [2019-12-01 19:03:50,099] {base_executor.py:132} DEBUG - Calling the <class 'airflow.executors.sequential_executor.SequentialExecutor'> sync method [2019-12-01 19:03:50,109] {cli_action_loggers.py:85} DEBUG - Calling callbacks: [] Traceback (most recent call last): File "/usr/local/bin/airflow", line 7, in <module> exec(compile(f.read(), __file__, 'exec')) File "/opt/airflow/airflow/bin/airflow", line 39, in <module> args.func(args) File "/opt/airflow/airflow/bin/cli.py", line 50, in command return func(*args, **kwargs) File "/opt/airflow/airflow/utils/cli.py", line 80, in wrapper return f(*args, **kwargs) File "/opt/airflow/airflow/cli/commands/dag_command.py", line 94, in dag_backfill run_backwards=args.run_backwards File "/opt/airflow/airflow/models/dag.py", line 1276, in run job.run() File "/opt/airflow/airflow/jobs/base_job.py", line 217, in run self._execute() File "/opt/airflow/airflow/utils/db.py", line 68, in wrapper return func(*args, **kwargs) File "/opt/airflow/airflow/jobs/backfill_job.py", line 773, in _execute raise AirflowException(err) airflow.exceptions.AirflowException: --------------------------------------------------- BackfillJob is deadlocked. These tasks have succeeded: set() These tasks are running: {} These tasks have failed: set() These tasks are skipped: set() These tasks are deadlocked: {<TaskInstance: demo_deadlock_backfill_issues.run 2019-01-02 00:00:00+00:00 [shutdown]>}{code}
> Please note that no attempt is made to start the sensor The task is blocked immediately. Probably because the task was previously canceled - CTRL+C.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)