Tasks can stick in the "queue" state due to the re-launch of DagFileProcessorManager

[avkirilishin]

Apache Airflow version

Other Airflow 2 version (please specify below)

What happened

Airflow 2.4.2 (but I suppose other versions too)

I noticed that sometimes some DAGs' and tasks' callbacks don't work. At that time, some tasks hung in the "queue" state and needed manual restarts. After some investigation, I determined that it happens during the re-launching of DagFileProcessorManager.

Here's what I think happens:

1. DagFileProcessorManager receives requests with callbacks (also failure callbacks) from the scheduler and puts them into the dict.
2. Then something goes wrong, and DagFileProcessorManager terminates.
3. DagFileProcessorAgent relaunches DagFileProcessorManager.
4. DagFileProcessorManager starts from scratch with the empty callback dict.
5. Due to the fact that tasks are marked as failed only in the DAG parsing process (after executing callbacks), some tasks with on_failure_callback or on_retry_callback can stay in the queue state. For example, I saw a sensor task that failed due to a network error (tasks could not be sent to the workers) and then stuck at the "queue" state.

What you think should happen instead

At least we can mark failed tasks with on_failure_callback and on_retry_callback at the scheduler, as we do for all other tasks. It prevents tasks from sticking.

But it would be great if other callbacks wouldn't be lost after the re-launching of DagFileProcessorManager.

How to reproduce

It's very hard to reproduce, but you can do something like that:

Add here this code

self.log.info("Callbacks to execute: %s", self._callback_to_execute)

Then run:

from datetime import datetime, timedelta
import pathlib
import time

from airflow.configuration import conf

from airflow.models import TaskInstance
from airflow.operators.bash import BashOperator
from airflow.models import DAG
from airflow.utils.state import State
from airflow.callbacks.callback_requests import TaskCallbackRequest
from airflow.models.taskinstance import SimpleTaskInstance
from airflow.dag_processing.manager import DagFileProcessorAgent

TEST_DAG_FOLDER = pathlib.Path(__file__) / 'dags'
test_dag_path = TEST_DAG_FOLDER / 'fake_dags.py'
async_mode = 'sqlite' not in conf.get('database', 'sql_alchemy_conn')
processor_agent = DagFileProcessorAgent(test_dag_path, -1, timedelta(days=365), [], False, async_mode)
processor_agent.start()

TI = TaskInstance(
    task=BashOperator(task_id="test", bash_command="true", dag=DAG(dag_id='id'),
                      start_date=datetime.now()),
    run_id="fake_run",
    state=State.RUNNING,
)
request = TaskCallbackRequest(
    full_filepath="filepath",
    simple_task_instance=SimpleTaskInstance.from_ti(ti=TI),
    processor_subdir='/test_dir',
    is_failure_callback=True,
)
processor_agent.get_callbacks_pipe().send(request)

time.sleep(5)
processor_agent._process.kill()

processor_agent.heartbeat()
time.sleep(5)
processor_agent._process.kill()

Then check the dag_processor_manager log to see that _callback_to_execute has had a callback and then became empty.

Operating System

CentOS Linux 7

Versions of Apache Airflow Providers

No response

Deployment

Virtualenv installation

Deployment details

No response

Anything else

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[potiuk]

Question: Are you talking about standalone DAG File Processor or the one that runs in Scheduler?

[avkirilishin]

Question: Are you talking about standalone DAG File Processor or the one that runs in Scheduler?

I am talking about standalone_dag_processor=False mode. Is that what you're asking?

But I looked into _fetch_callbacks (used for standalone_dag_processor=True mode) and there is, I suppose, the same problem. We add callbacks to the queue and immediately remove them from the database without checking for successful completion.

[potiuk]

Question: Are you talking about standalone DAG File Processor or the one that runs in Scheduler?

I am talking about standalone_dag_processor=False mode. Is that what you're asking?

yes. I was just checking.

But I looked into _fetch_callbacks (used for standalone_dag_processor=True mode) and there is, I suppose, the same problem. We add callbacks to the queue and immediately remove them from the database without checking for successful completion.

Yeah I think this edge case needs to be investigated and fixed.

[avkirilishin]

So, can we divide this problem into two parts?

1. Consider marking failed tasks with on_failure_callback and on_retry_callback at the scheduler, as we do for all other tasks. It's a more critical part because it can lead to tasks sticking.
2. Investigate and fix potential callback losses. It's a much more complicated part.

[potiuk]

Could be. Possibly someone who knows that part better can chime in and comment.

[RNHTTR]

If this is impacting you, you can use stalled_task_timeout if you're using CeleryExecutor or worker_pods_pending_timeout if you're using KubernetesExecutor. Or upgrade to 2.6+ and use task_queued_timeout regardless of executor.