Deferrable ExternalTaskSensor times out even if DAG is running

[ecodina]

Apache Airflow version

2.7.1

What happened

If a DAG (dag1) is running and another DAG (dag2) has an ExternalTaskSensor (task-externalsensor) that checks a task on dag1, task-externalsensor will fail unless dag1's task finishes in under 60 seconds.

After checking the code, I believe the while block in the trigger should be split in two, as follows:

# Wait for the dag to start execution
# Maybe the timeout should be configurable?
while True:
    try:
        delta = utcnow() - self.trigger_start_time
        if delta.total_seconds() < self._timeout_sec:
            # mypy confuses typing here
            if await self.count_running_dags() == 0:  # type: ignore[call-arg]
                self.log.info("Waiting for DAG to start execution...")
                await asyncio.sleep(self.poll_interval)
            else:
                break
        else:
            yield TriggerEvent({"status": "timeout"})
            return
    except Exception:
        yield TriggerEvent({"status": "failed"})
        return

# Wait for the external task to finish
# Maybe out a timeout here?
while True:
    try:
        if await self.count_tasks() == len(self.execution_dates):  # type: ignore[call-arg]
            yield TriggerEvent({"status": "success"})
            return
        self.log.info("Task is still running, sleeping for %s seconds...", self.poll_interval)
        await asyncio.sleep(self.poll_interval)
    except Exception:
        yield TriggerEvent({"status": "failed"})
        return

I tested this code on my local machine and it worked as expected with dag1 running for 120 seconds.

Maybe @bkossakowska could check this out?

What you think should happen instead

No response

How to reproduce

1. Create dag1 which contains a simple task (sleep for > 60 seconds)
2. Create dag2 which contains an ExternalTaskSensor checking for dag1's task and runs in deferrable mode
3. Execute them at the same time. As soon as dag2's trigger has spent 60 seconds, it will fail, even if dag1's task is still running.

Operating System

Ubuntu 22.04.3 LTS

Versions of Apache Airflow Providers

No response

Deployment

Virtualenv installation

Deployment details

No response

Anything else

I'd love to submit a PR, but have no available time to properly do so or the knowledge to make sure the proposed solution actually works in all cases.

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

[potiuk]

I'd love to submit a PR, but have no available time to properly do so or the knowledge to make sure the proposed solution actually works in all cases.

then it will be waiting for someone to pick it up. But you might find it still here when you come-back ... so maybe it will be you eventually.

[ecodina]

I'd love to submit a PR, but have no available time to properly do so or the knowledge to make sure the proposed solution actually works in all cases.

then it will be waiting for someone to pick it up. But you might find it still here when you come-back ... so maybe it will be you eventually.

I contributed once last year to Airflow. However, I now have a job and I'm a PhD student, which is why I commented I have no available time. I spent as much time as I could debugging the issue to help whoever picks it up!

I really look forward to meeting you next week in the summit to thank you in person for all the help you've provided us the last past months!

[raffifu]

Hiii, can i work on this issue?

[raffifu]

I can reproduce the issue. I think it's because the default timeout value set to 60 seconds on the trigger. Does the trigger of ExternalTaskSensor should wait for the dependent dags to complete even it takes more than the timeout?

[ecodina]

I believe so. I don't think it's safe to assume how long a running DAG will take to finish (maybe it's doing I/O operations over an unreliable network). That's why I proposed to split the block: first, wait for the DAG to be running (where a timeout I feel it's appropriate) and then actually wait for the DAG to finish however long it takes.

Oh, and I appreciate you taking this issue!

[raffifu]

Got it, after reading the comment on the code again. Timeout only happen if the dependent task is not running in the period times. If the task is on the running state, ExternalTaskSensor must wait until the task complete. I will try the purposed solution and test on the other cases

[raffifu]

@potiuk is the deferrable mode should behave the same as non deferable mode on ExternalTaskSensor? Because i found that the deferrable will not fail if the external_task_id failed. I think it's because there's no checking about the state of task or dags on the run method.
This error happen when i'm trying to implement the proposed solution from @ecodina. But, even if we're using the current code, it will fail because of the timeout not because the external_task failed (it will confused the user i think)

  while True:
      delta = utcnow() - self.trigger_start_time
      if delta.total_seconds() < self._timeout_sec:
          if await self.count_running_dags() == 0:  # type: ignore[call-arg]
              self.log.info("Waiting for DAG to start execution...")
              await asyncio.sleep(self.poll_interval)
          else:
              break
      else:
          yield TriggerEvent({"status": "timeout"})
          return

  # Wait Task to success
  while True:
      if await self.count_tasks() == len(self.execution_dates):  # type: ignore[call-arg]
          yield TriggerEvent({"status": "success"})
          return

      self.log.info("Task is still running, sleeping for %s seconds...", self.poll_interval)
      await asyncio.sleep(self.poll_interval)

Current code:

while True:
    delta = utcnow() - self.trigger_start_time
    if delta.total_seconds() < self._timeout_sec:
        # mypy confuses typing here
        if await self.count_running_dags() == 0:  # type: ignore[call-arg]
            self.log.info("Waiting for DAG to start execution...")
            await asyncio.sleep(self.poll_interval)
    else:
        yield TriggerEvent({"status": "timeout"}) # <---- This will triggered if the running task is suddenly failed
        return
    # mypy confuses typing here
    if await self.count_tasks() == len(self.execution_dates):  # type: ignore[call-arg]
        yield TriggerEvent({"status": "success"})
        return
    self.log.info("Task is still running, sleeping for %s seconds...", self.poll_interval)
    await asyncio.sleep(self.poll_interval)