Skip to content

Reschedule sensor doesn't apply execution timeout #32512

Closed
Closed
Reschedule sensor doesn't apply execution timeout#32512
Labels
area:corekind:bugThis is a clearly a bugneeds-triagelabel for new issues that we didn't triage yet

Description

@AlexisBRENON
AlexisBRENON
opened on Jul 11, 2023

Apache Airflow version

2.6.2

What happened

I created a sensor in reschedule mode with a given execution timeout.
However, the sensor never changes to failed test, even after the execution timeout passed.

What you think should happen instead

I expect the sensor to be poked a given number of times and then changing to failed state while the execution timeout elapsed.

How to reproduce

Define a DAG with a single sensor, checking for an innexistent file.
The sensor should poke every 30s and timeout after 1 minutes.

import datetime
from airflow.models.dag import DAG
from airflow.sensors.filesystem import FileSensor

with DAG(dag_id="foo",
         schedule_interval="@once",
         start_date=datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=1),
         catchup=False) as dag:
    FileSensor(task_id="wait_for_file",
               filepath="/tmp/temporary_file_for_testing",
               poke_interval=30,
               mode="reschedule",
               execution_timeout=datetime.timedelta(minutes=1))

Operating System

Arch Linux

Versions of Apache Airflow Providers

No response

Deployment

Other Docker-based deployment

Deployment details

docker run --rm -it -v /tmp/dags:/opt/airflow/dags -p 8080:8080 apache/airflow:2.6.2-python3.11 airflow standalone

Similar behavior encountered in my production deployment based on Helm chart.

Anything else

Logs of the task. You can see that the log covers more than 1 minute.

[2023-07-11, 08:26:38 UTC] {taskinstance.py:1103} INFO - Dependencies all met for dep_context=non-requeueable deps ti=<TaskInstance: foo.wait_for_file scheduled__2023-07-10T08:26:07.521239+00:00 [queued]>
[2023-07-11, 08:26:38 UTC] {taskinstance.py:1103} INFO - Dependencies all met for dep_context=requeueable deps ti=<TaskInstance: foo.wait_for_file scheduled__2023-07-10T08:26:07.521239+00:00 [queued]>
[2023-07-11, 08:26:38 UTC] {taskinstance.py:1308} INFO - Starting attempt 1 of 1
[2023-07-11, 08:26:38 UTC] {taskinstance.py:1327} INFO - Executing <Task(FileSensor): wait_for_file> on 2023-07-10 08:26:07.521239+00:00
[2023-07-11, 08:26:38 UTC] {standard_task_runner.py:57} INFO - Started process 55 to run task
[2023-07-11, 08:26:38 UTC] {standard_task_runner.py:84} INFO - Running: ['airflow', 'tasks', 'run', 'foo', 'wait_for_file', 'scheduled__2023-07-10T08:26:07.521239+00:00', '--job-id', '3', '--raw', '--subdir', 'DAGS_FOLDER/test.py', '--cfg-path', '/tmp/tmpkxpv9y3q']
[2023-07-11, 08:26:38 UTC] {standard_task_runner.py:85} INFO - Job 3: Subtask wait_for_file
[2023-07-11, 08:26:38 UTC] {task_command.py:410} INFO - Running <TaskInstance: foo.wait_for_file scheduled__2023-07-10T08:26:07.521239+00:00 [running]> on host 686510c65d84
[2023-07-11, 08:26:38 UTC] {taskinstance.py:1545} INFO - Exporting env vars: AIRFLOW_CTX_DAG_OWNER='airflow' AIRFLOW_CTX_DAG_ID='foo' AIRFLOW_CTX_TASK_ID='wait_for_file' AIRFLOW_CTX_EXECUTION_DATE='2023-07-10T08:26:07.521239+00:00' AIRFLOW_CTX_TRY_NUMBER='1' AIRFLOW_CTX_DAG_RUN_ID='scheduled__2023-07-10T08:26:07.521239+00:00'
[2023-07-11, 08:26:38 UTC] {base.py:73} INFO - Using connection ID 'fs_default' for task execution.
[2023-07-11, 08:26:38 UTC] {filesystem.py:64} INFO - Poking for file /tmp/temporary_file_for_testing
[2023-07-11, 08:26:38 UTC] {taskinstance.py:1784} INFO - Rescheduling task, marking task as UP_FOR_RESCHEDULE
[2023-07-11, 08:26:38 UTC] {local_task_job_runner.py:225} INFO - Task exited with return code 0
[2023-07-11, 08:26:38 UTC] {taskinstance.py:2653} INFO - 0 downstream tasks scheduled from follow-on schedule check
[2023-07-11, 08:27:10 UTC] {taskinstance.py:1103} INFO - Dependencies all met for dep_context=non-requeueable deps ti=<TaskInstance: foo.wait_for_file scheduled__2023-07-10T08:26:07.521239+00:00 [queued]>
[2023-07-11, 08:27:10 UTC] {taskinstance.py:1103} INFO - Dependencies all met for dep_context=requeueable deps ti=<TaskInstance: foo.wait_for_file scheduled__2023-07-10T08:26:07.521239+00:00 [queued]>
[2023-07-11, 08:27:10 UTC] {taskinstance.py:1308} INFO - Starting attempt 1 of 1
[2023-07-11, 08:27:11 UTC] {taskinstance.py:1327} INFO - Executing <Task(FileSensor): wait_for_file> on 2023-07-10 08:26:07.521239+00:00
[2023-07-11, 08:27:11 UTC] {standard_task_runner.py:57} INFO - Started process 58 to run task
[2023-07-11, 08:27:11 UTC] {standard_task_runner.py:84} INFO - Running: ['airflow', 'tasks', 'run', 'foo', 'wait_for_file', 'scheduled__2023-07-10T08:26:07.521239+00:00', '--job-id', '4', '--raw', '--subdir', 'DAGS_FOLDER/test.py', '--cfg-path', '/tmp/tmpaq2r2iaf']
[2023-07-11, 08:27:11 UTC] {standard_task_runner.py:85} INFO - Job 4: Subtask wait_for_file
[2023-07-11, 08:27:11 UTC] {task_command.py:410} INFO - Running <TaskInstance: foo.wait_for_file scheduled__2023-07-10T08:26:07.521239+00:00 [running]> on host 686510c65d84
[2023-07-11, 08:27:11 UTC] {taskinstance.py:1545} INFO - Exporting env vars: AIRFLOW_CTX_DAG_OWNER='airflow' AIRFLOW_CTX_DAG_ID='foo' AIRFLOW_CTX_TASK_ID='wait_for_file' AIRFLOW_CTX_EXECUTION_DATE='2023-07-10T08:26:07.521239+00:00' AIRFLOW_CTX_TRY_NUMBER='1' AIRFLOW_CTX_DAG_RUN_ID='scheduled__2023-07-10T08:26:07.521239+00:00'
[2023-07-11, 08:27:11 UTC] {base.py:73} INFO - Using connection ID 'fs_default' for task execution.
[2023-07-11, 08:27:11 UTC] {filesystem.py:64} INFO - Poking for file /tmp/temporary_file_for_testing
[2023-07-11, 08:27:11 UTC] {taskinstance.py:1784} INFO - Rescheduling task, marking task as UP_FOR_RESCHEDULE
[2023-07-11, 08:27:11 UTC] {local_task_job_runner.py:225} INFO - Task exited with return code 0
[2023-07-11, 08:27:11 UTC] {taskinstance.py:2653} INFO - 0 downstream tasks scheduled from follow-on schedule check
[2023-07-11, 08:27:43 UTC] {taskinstance.py:1103} INFO - Dependencies all met for dep_context=non-requeueable deps ti=<TaskInstance: foo.wait_for_file scheduled__2023-07-10T08:26:07.521239+00:00 [queued]>
[2023-07-11, 08:27:43 UTC] {taskinstance.py:1103} INFO - Dependencies all met for dep_context=requeueable deps ti=<TaskInstance: foo.wait_for_file scheduled__2023-07-10T08:26:07.521239+00:00 [queued]>
[2023-07-11, 08:27:43 UTC] {taskinstance.py:1308} INFO - Starting attempt 1 of 1
[2023-07-11, 08:27:43 UTC] {taskinstance.py:1327} INFO - Executing <Task(FileSensor): wait_for_file> on 2023-07-10 08:26:07.521239+00:00
[2023-07-11, 08:27:43 UTC] {standard_task_runner.py:57} INFO - Started process 61 to run task
[2023-07-11, 08:27:43 UTC] {standard_task_runner.py:84} INFO - Running: ['airflow', 'tasks', 'run', 'foo', 'wait_for_file', 'scheduled__2023-07-10T08:26:07.521239+00:00', '--job-id', '5', '--raw', '--subdir', 'DAGS_FOLDER/test.py', '--cfg-path', '/tmp/tmpn89kv31n']
[2023-07-11, 08:27:43 UTC] {standard_task_runner.py:85} INFO - Job 5: Subtask wait_for_file
[2023-07-11, 08:27:43 UTC] {task_command.py:410} INFO - Running <TaskInstance: foo.wait_for_file scheduled__2023-07-10T08:26:07.521239+00:00 [running]> on host 686510c65d84
[2023-07-11, 08:27:44 UTC] {taskinstance.py:1545} INFO - Exporting env vars: AIRFLOW_CTX_DAG_OWNER='airflow' AIRFLOW_CTX_DAG_ID='foo' AIRFLOW_CTX_TASK_ID='wait_for_file' AIRFLOW_CTX_EXECUTION_DATE='2023-07-10T08:26:07.521239+00:00' AIRFLOW_CTX_TRY_NUMBER='1' AIRFLOW_CTX_DAG_RUN_ID='scheduled__2023-07-10T08:26:07.521239+00:00'
[2023-07-11, 08:27:44 UTC] {base.py:73} INFO - Using connection ID 'fs_default' for task execution.
[2023-07-11, 08:27:44 UTC] {filesystem.py:64} INFO - Poking for file /tmp/temporary_file_for_testing
[2023-07-11, 08:27:44 UTC] {taskinstance.py:1784} INFO - Rescheduling task, marking task as UP_FOR_RESCHEDULE
[2023-07-11, 08:27:44 UTC] {local_task_job_runner.py:225} INFO - Task exited with return code 0
[2023-07-11, 08:27:44 UTC] {taskinstance.py:2653} INFO - 0 downstream tasks scheduled from follow-on schedule check
[2023-07-11, 08:28:16 UTC] {taskinstance.py:1103} INFO - Dependencies all met for dep_context=non-requeueable deps ti=<TaskInstance: foo.wait_for_file scheduled__2023-07-10T08:26:07.521239+00:00 [queued]>
[2023-07-11, 08:28:16 UTC] {taskinstance.py:1103} INFO - Dependencies all met for dep_context=requeueable deps ti=<TaskInstance: foo.wait_for_file scheduled__2023-07-10T08:26:07.521239+00:00 [queued]>
[2023-07-11, 08:28:16 UTC] {taskinstance.py:1308} INFO - Starting attempt 1 of 1
[2023-07-11, 08:28:16 UTC] {taskinstance.py:1327} INFO - Executing <Task(FileSensor): wait_for_file> on 2023-07-10 08:26:07.521239+00:00
[2023-07-11, 08:28:16 UTC] {standard_task_runner.py:57} INFO - Started process 64 to run task
[2023-07-11, 08:28:16 UTC] {standard_task_runner.py:84} INFO - Running: ['airflow', 'tasks', 'run', 'foo', 'wait_for_file', 'scheduled__2023-07-10T08:26:07.521239+00:00', '--job-id', '6', '--raw', '--subdir', 'DAGS_FOLDER/test.py', '--cfg-path', '/tmp/tmpartoz87c']
[2023-07-11, 08:28:16 UTC] {standard_task_runner.py:85} INFO - Job 6: Subtask wait_for_file
[2023-07-11, 08:28:16 UTC] {task_command.py:410} INFO - Running <TaskInstance: foo.wait_for_file scheduled__2023-07-10T08:26:07.521239+00:00 [running]> on host 686510c65d84
[2023-07-11, 08:28:16 UTC] {taskinstance.py:1545} INFO - Exporting env vars: AIRFLOW_CTX_DAG_OWNER='airflow' AIRFLOW_CTX_DAG_ID='foo' AIRFLOW_CTX_TASK_ID='wait_for_file' AIRFLOW_CTX_EXECUTION_DATE='2023-07-10T08:26:07.521239+00:00' AIRFLOW_CTX_TRY_NUMBER='1' AIRFLOW_CTX_DAG_RUN_ID='scheduled__2023-07-10T08:26:07.521239+00:00'
[2023-07-11, 08:28:16 UTC] {base.py:73} INFO - Using connection ID 'fs_default' for task execution.
[2023-07-11, 08:28:16 UTC] {filesystem.py:64} INFO - Poking for file /tmp/temporary_file_for_testing
[2023-07-11, 08:28:16 UTC] {taskinstance.py:1784} INFO - Rescheduling task, marking task as UP_FOR_RESCHEDULE
[2023-07-11, 08:28:16 UTC] {local_task_job_runner.py:225} INFO - Task exited with return code 0
[2023-07-11, 08:28:16 UTC] {taskinstance.py:2653} INFO - 0 downstream tasks scheduled from follow-on schedule check
[2023-07-11, 08:28:48 UTC] {taskinstance.py:1103} INFO - Dependencies all met for dep_context=non-requeueable deps ti=<TaskInstance: foo.wait_for_file scheduled__2023-07-10T08:26:07.521239+00:00 [queued]>
[2023-07-11, 08:28:48 UTC] {taskinstance.py:1103} INFO - Dependencies all met for dep_context=requeueable deps ti=<TaskInstance: foo.wait_for_file scheduled__2023-07-10T08:26:07.521239+00:00 [queued]>
[2023-07-11, 08:28:48 UTC] {taskinstance.py:1308} INFO - Starting attempt 1 of 1
[2023-07-11, 08:28:48 UTC] {taskinstance.py:1327} INFO - Executing <Task(FileSensor): wait_for_file> on 2023-07-10 08:26:07.521239+00:00
[2023-07-11, 08:28:48 UTC] {standard_task_runner.py:57} INFO - Started process 67 to run task
[2023-07-11, 08:28:48 UTC] {standard_task_runner.py:84} INFO - Running: ['airflow', 'tasks', 'run', 'foo', 'wait_for_file', 'scheduled__2023-07-10T08:26:07.521239+00:00', '--job-id', '7', '--raw', '--subdir', 'DAGS_FOLDER/test.py', '--cfg-path', '/tmp/tmplriw9lrd']
[2023-07-11, 08:28:48 UTC] {standard_task_runner.py:85} INFO - Job 7: Subtask wait_for_file
[2023-07-11, 08:28:48 UTC] {task_command.py:410} INFO - Running <TaskInstance: foo.wait_for_file scheduled__2023-07-10T08:26:07.521239+00:00 [running]> on host 686510c65d84
[2023-07-11, 08:28:48 UTC] {taskinstance.py:1545} INFO - Exporting env vars: AIRFLOW_CTX_DAG_OWNER='airflow' AIRFLOW_CTX_DAG_ID='foo' AIRFLOW_CTX_TASK_ID='wait_for_file' AIRFLOW_CTX_EXECUTION_DATE='2023-07-10T08:26:07.521239+00:00' AIRFLOW_CTX_TRY_NUMBER='1' AIRFLOW_CTX_DAG_RUN_ID='scheduled__2023-07-10T08:26:07.521239+00:00'
[2023-07-11, 08:28:48 UTC] {base.py:73} INFO - Using connection ID 'fs_default' for task execution.
[2023-07-11, 08:28:48 UTC] {filesystem.py:64} INFO - Poking for file /tmp/temporary_file_for_testing
[2023-07-11, 08:28:48 UTC] {taskinstance.py:1784} INFO - Rescheduling task, marking task as UP_FOR_RESCHEDULE
[2023-07-11, 08:28:48 UTC] {local_task_job_runner.py:225} INFO - Task exited with return code 0
[2023-07-11, 08:28:48 UTC] {taskinstance.py:2653} INFO - 0 downstream tasks scheduled from follow-on schedule check
...

Are you willing to submit PR?

  • Yes I am willing to submit a PR!

Code of Conduct

Metadata

Metadata

Assignees

No one assigned

    Labels

    area:corekind:bugThis is a clearly a bugneeds-triagelabel for new issues that we didn't triage yet

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions