Celery tasks stuck queued. Clearing task fails rescheduling

[hterik]

Apache Airflow version

Other Airflow 2 version (please specify below)

What happened

Task set to run in celery queue was queued but never started on the worker.
Stuck in state PENDING even if celery queue had spare capacity.

[2023-11-24T11:41:49.714+0000] {scheduler_job_runner.py:414} INFO - 1 tasks up for execution:
    <TaskInstance: MYDAGID.MYTASKID MYRUNID [scheduled]>
[2023-11-24T11:41:49.714+0000] {scheduler_job_runner.py:593} INFO - Setting the following tasks to queued state:
    <TaskInstance: MYDAGID.MYTASKID MYRUNID [scheduled]>
[2023-11-24T11:41:49.716+0000] {scheduler_job_runner.py:636} INFO - Sending TaskInstanceKey(dag_id='MYDAGID', task_id='MYTASKID', run_id='MYRUNID', try_number=1, map_index=-1) to executor with priority 1 and queue MYQUEUE
[2023-11-24T11:41:49.716+0000] {base_executor.py:144} INFO - Adding to queue: ['airflow', 'tasks', 'run', 'MYDAGID', 'MYTASKID', 'MYRUNID', '--local', '--subdir', 'DAGS_FOLDER/dag_xxx.py']
[2023-11-24T11:41:49.766+0000] {scheduler_job_runner.py:686} INFO - Received executor event with state queued for task instance TaskInstanceKey(dag_id='MYDAGID', task_id='MYTASKID', run_id='MYRUNID', try_number=1, map_index=-1)
[2023-11-24T11:41:49.769+0000] {scheduler_job_runner.py:713} INFO - Setting external_id for <TaskInstance: MYDAGID.MYTASKID MYRUNID [queued]> to 5502f29e-4f17-40e3-b91e-519c3fb4606b

[2023-11-24T14:21:45.680+0000] {celery_executor.py:439} INFO - Adopted the following 22 tasks from a dead executor
    ...
    <TaskInstance: MYDAGID.MYTASKID MYRUNID [queued]> in state PENDING
[2023-11-24T14:48:24.426+0000] {celery_executor.py:439} INFO - Adopted the following 25 tasks from a dead executor
    ...
    <TaskInstance: MYDAGID.MYTASKID MYRUNID [queued]> in state PENDING

At this point, if i introspect the Celery queue, it does not contain any item called 5502f29e-4f17-40e3-b91e-519c3fb4606b

After clicking on Clear task in Airflow UI, scheduler continues failing to schedule the task like follows:

[2023-11-24T15:44:58.737+0000] {scheduler_job_runner.py:414} INFO - 1 tasks up for execution:
    <TaskInstance: MYDAGID.MYTASKID MYRUNID [scheduled]>
[2023-11-24T15:44:58.738+0000] {scheduler_job_runner.py:593} INFO - Setting the following tasks to queued state:
    <TaskInstance: MYDAGID.MYTASKID MYRUNID [scheduled]>
[2023-11-24T15:44:58.739+0000] {scheduler_job_runner.py:636} INFO - Sending TaskInstanceKey(dag_id='MYDAGID', task_id='MYTASKID', run_id='MYRUNID', try_number=1, map_index=-1) to executor with priority 1 and queue MYQUEUE
[2023-11-24T15:44:58.739+0000] {base_executor.py:144} INFO - Adding to queue: ['airflow', 'tasks', 'run', 'MYDAGID', 'MYTASKID', 'MYRUNID', '--local', '--subdir', 'DAGS_FOLDER/dag_xxx.py']

[2023-11-24T15:44:58.741+0000] {base_executor.py:277} INFO - queued but still running; attempt=1 task=TaskInstanceKey(dag_id='MYDAGID', task_id='MYTASKID', run_id='MYRUNID', try_number=1, map_index=-1)
[2023-11-24T15:44:59.045+0000] {base_executor.py:277} INFO - queued but still running; attempt=2 task=TaskInstanceKey(dag_id='MYDAGID', task_id='MYTASKID', run_id='MYRUNID', try_number=1, map_index=-1)
[2023-11-24T15:44:59.368+0000] {base_executor.py:277} INFO - queued but still running; attempt=3 task=TaskInstanceKey(dag_id='MYDAGID', task_id='MYTASKID', run_id='MYRUNID', try_number=1, map_index=-1)
[2023-11-24T15:45:00.669+0000] {base_executor.py:277} INFO - queued but still running; attempt=4 task=TaskInstanceKey(dag_id='MYDAGID', task_id='MYTASKID', run_id='MYRUNID', try_number=1, map_index=-1)
[2023-11-24T15:45:01.161+0000] {base_executor.py:277} INFO - queued but still running; attempt=5 task=TaskInstanceKey(dag_id='MYDAGID', task_id='MYTASKID', run_id='MYRUNID', try_number=1, map_index=-1)
[2023-11-24T15:45:02.069+0000] {base_executor.py:277} INFO - queued but still running; attempt=6 task=TaskInstanceKey(dag_id='MYDAGID', task_id='MYTASKID', run_id='MYRUNID', try_number=1, map_index=-1)
[2023-11-24T15:45:03.339+0000] {base_executor.py:277} INFO - queued but still running; attempt=7 task=TaskInstanceKey(dag_id='MYDAGID', task_id='MYTASKID', run_id='MYRUNID', try_number=1, map_index=-1)
[2023-11-24T15:45:04.754+0000] {base_executor.py:277} INFO - queued but still running; attempt=8 task=TaskInstanceKey(dag_id='MYDAGID', task_id='MYTASKID', run_id='MYRUNID', try_number=1, map_index=-1)
[2023-11-24T15:45:06.083+0000] {base_executor.py:277} INFO - queued but still running; attempt=9 task=TaskInstanceKey(dag_id='MYDAGID', task_id='MYTASKID', run_id='MYRUNID', try_number=1, map_index=-1)
[2023-11-24T15:45:07.378+0000] {base_executor.py:277} INFO - queued but still running; attempt=10 task=TaskInstanceKey(dag_id='MYDAGID', task_id='MYTASKID', run_id='MYRUNID', try_number=1, map_index=-1)
[2023-11-24T15:45:08.629+0000] {base_executor.py:277} INFO - queued but still running; attempt=11 task=TaskInstanceKey(dag_id='MYDAGID', task_id='MYTASKID', run_id='MYRUNID', try_number=1, map_index=-1)
[2023-11-24T15:45:09.675+0000] {base_executor.py:277} INFO - queued but still running; attempt=12 task=TaskInstanceKey(dag_id='MYDAGID', task_id='MYTASKID', run_id='MYRUNID', try_number=1, map_index=-1)
[2023-11-24T15:45:10.983+0000] {base_executor.py:280} ERROR - could not queue task TaskInstanceKey(dag_id='MYDAGID', task_id='MYTASKID', run_id='MYRUNID', try_number=1, map_index=-1) (still running after 12 attempts)

I tried Clear Task twice and it gave exactly the same result as above again.
At this point, restart airflow-scheduler.

[2023-11-24T16:17:57.328+0000] {scheduler_job_runner.py:1650} INFO - Reset the following 1 orphaned TaskInstances:
    <TaskInstance: MYDAGID.MYTASKID MYRUNID [queued]>
[2023-11-24T16:17:57.876+0000] {scheduler_job_runner.py:414} INFO - 2 tasks up for execution:
    <TaskInstance: MYDAGID.MYTASKID MYRUNID [scheduled]>
    <TaskInstance: autobump_platform_master.first_seen_commit autobump_platform_master-412983e6bd [scheduled]>
[2023-11-24T16:17:57.877+0000] {scheduler_job_runner.py:593} INFO - Setting the following tasks to queued state:
    <TaskInstance: MYDAGID.MYTASKID MYRUNID [scheduled]>
    <TaskInstance: autobump_platform_master.first_seen_commit autobump_platform_master-412983e6bd [scheduled]>
[2023-11-24T16:17:57.880+0000] {scheduler_job_runner.py:636} INFO - Sending TaskInstanceKey(dag_id='MYDAGID', task_id='MYTASKID', run_id='MYRUNID', try_number=1, map_index=-1) to executor with priority 1 and queue MYQUEUE
[2023-11-24T16:17:57.880+0000] {base_executor.py:144} INFO - Adding to queue: ['airflow', 'tasks', 'run', 'MYDAGID', 'MYTASKID', 'MYRUNID', '--local', '--subdir', 'DAGS_FOLDER/dag_xxx.py']
[2023-11-24T16:17:58.377+0000] {scheduler_job_runner.py:686} INFO - Received executor event with state queued for task instance TaskInstanceKey(dag_id='MYDAGID', task_id='MYTASKID', run_id='MYRUNID', try_number=1, map_index=-1)
[2023-11-24T16:17:58.387+0000] {scheduler_job_runner.py:713} INFO - Setting external_id for <TaskInstance: MYDAGID.MYTASKID MYRUNID [queued]> to 7f00daa5-1ff9-48d9-9d3d-dacf871c27d6
[2023-11-24T16:19:16.805+0000] {celery_executor.py:439} INFO - Adopted the following 18 tasks from a dead executor
    <TaskInstance: MYDAGID.MYTASKID MYRUNID [running]> in state STARTED
	...

Now it works. Notice new external_id was given, 7f00daa5-1ff9-48d9-9d3d-dacf871c27d6, which i can observe in the celery queue.

What you think should happen instead

1. Celery tasks should not get stuck
2. Clicking Clear task should reschedule stuck tasks more reliably
3. Restarting airflow-scheduler should not be needed to reschedule stuck tasks.

How to reproduce

Dunno :(

Operating System

Debian GNU/Linux 11 (bullseye)

Versions of Apache Airflow Providers

apache-airflow==2.7.1
apache-airflow-client==2.1.0
apache-airflow-providers-celery==3.3.3
apache-airflow-providers-cncf-kubernetes==7.5.0

Deployment

Official Apache Airflow Helm Chart

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

[RNHTTR]

Did this just happen once or does it happen intermittently? Frequently?

[hterik]

Anecdotally, I've personally seen it happen at least three times in 2 weeks. I don't have any other detailed analysis of the actual frequency.

[github-actions[bot]]

This issue has been automatically marked as stale because it has been open for 14 days with no response from the author. It will be closed in next 7 days if no further activity occurs from the issue author.

[hterik]

unstale

[alex-astronomer]

I think that I'm having the same issue here:

Celery Executor

There was a task that was stuck in a queued state. The scheduler logs state that the task was first queued at 12:55. At this point it's assigned an external_id

[2023-11-30T12:55:16.877+0000] {scheduler_job_runner.py:411} INFO - 1 tasks up for execution:
<TaskInstance: DAG_A.TG_1.TASK_1 manual__2023-11-30T12:54:48.282524+00:00 [scheduled]>
[2023-11-30T12:55:16.878+0000] {scheduler_job_runner.py:587} INFO - Setting the following tasks to queued state:
<TaskInstance: DAG_A.TG_1.TASK_1 manual__2023-11-30T12:54:48.282524+00:00 [scheduled]>
[2023-11-30T12:55:16.951+0000] {scheduler_job_runner.py:625} INFO - Sending TaskInstanceKey(dag_id='DAG_A', task_id='TG_1.TASK_1', run_id='manual__2023-11-30T12:54:48.282524+00:00', try_number=1, map_index=-1) to executor with priority 2 and queue default
[2023-11-30T12:55:16.951+0000] {base_executor.py:147} INFO - Adding to queue: ['airflow', 'tasks', 'run', 'DAG_A', 'TG_1.TASK_1', 'manual__2023-11-30T12:54:48.282524+00:00', '--local', '--subdir', 'DAGS_FOLDER/.../DAG_FILE.py']
[2023-11-30T12:55:17.167+0000] {scheduler_job_runner.py:677} INFO - Received executor event with state queued for task instance TaskInstanceKey(dag_id='DAG_A', task_id='TG_1.TASK_1', run_id='manual__2023-11-30T12:54:48.282524+00:00', try_number=1, map_index=-1)
[2023-11-30T12:55:17.256+0000] {scheduler_job_runner.py:703} INFO - Setting external_id for <TaskInstance: DAG_A.TG_1.TASK_1 manual__2023-11-30T12:54:48.282524+00:00 [queued]> to 96c9f186-97a4-4bbf-a826-cffbd5ba9b98

At this point, the task is queued and has that external_id assigned. My understanding is that, at this point, it's in the executor queue but hasn't made it to the worker yet for execution.

From here, the task is cleared from the UI since it was appearing stuck in queued for a long period of time, however when the task was cleared and it tried to re-queue the logs showed that the task was in a running state.

[2023-11-30T13:04:23.267+0000] {scheduler_job_runner.py:411} INFO - 2 tasks up for execution:
...
<TaskInstance: DAG_A.TG_1.TASK_1 manual__2023-11-30T12:54:48.282524+00:00 [scheduled]>
[2023-11-30T13:04:23.268+0000] {scheduler_job_runner.py:587} INFO - Setting the following tasks to queued state:
...
<TaskInstance: DAG_A.TG_1.TASK_1 manual__2023-11-30T12:54:48.282524+00:00 [scheduled]>
[2023-11-30T13:04:23.272+0000] {scheduler_job_runner.py:625} INFO - Sending TaskInstanceKey(dag_id='DAG_A', task_id='TG_1.TASK_1', run_id='manual__2023-11-30T12:54:48.282524+00:00', try_number=1, map_index=-1) to executor with priority 2 and queue default
[2023-11-30T13:04:23.272+0000] {base_executor.py:147} INFO - Adding to queue: ['airflow', 'tasks', 'run', 'DAG_A', 'TG_1.TASK_1', 'manual__2023-11-30T12:54:48.282524+00:00', '--local', '--subdir', 'DAGS_FOLDER/fund_admin/DAG_FILE.py']
[2023-11-30T13:04:23.275+0000] {base_executor.py:269} INFO - queued but still running; attempt=1 task=TaskInstanceKey(dag_id='DAG_A', task_id='TG_1.TASK_1', run_id='manual__2023-11-30T12:54:48.282524+00:00', try_number=1, map_index=-1)
[2023-11-30T13:04:28.068+0000] {base_executor.py:269} INFO - queued but still running;  attempt=2 task=TaskInstanceKey(dag_id='DAG_A', task_id='TG_1.TASK_1', run_id='manual__2023-11-30T12:54:48.282524+00:00', try_number=1, map_index=-1)
[2023-11-30T13:04:32.571+0000] {base_executor.py:269} INFO - queued but still running; attempt=3 task=TaskInstanceKey(dag_id='DAG_A', task_id='TG_1.TASK_1', run_id='manual__2023-11-30T12:54:48.282524+00:00', try_number=1, map_index=-1)
[2023-11-30T13:04:37.755+0000] {base_executor.py:269} INFO - queued but still running; attempt=4 task=TaskInstanceKey(dag_id='DAG_A', task_id='TG_1.TASK_1', run_id='manual__2023-11-30T12:54:48.282524+00:00', try_number=1, map_index=-1)
[2023-11-30T13:04:44.255+0000] {base_executor.py:272} ERROR - could not queue task TaskInstanceKey(dag_id='DAG_A', task_id='TG_1.TASK_1', run_id='manual__2023-11-30T12:54:48.282524+00:00', try_number=1, map_index=-1) (still running after 4 attempts)

This repeats a few times until the task finally runs. The task finally runs around 13:28 (~30 minutes from when it was first queued). During that time, a new worker was spawned (KEDA and cluster-autoscaler) and the task finally runs.

Let me know if you have further questions!

[RNHTTR]

Until this can be reproduced, I'm going to convert this to a discussion. Once someone has a reliable reproduction, it can be converted back to an Issue.

[flinz]

I am observing the same on a LocalExecutor on 2.9.1 deployed on kubernetes via helm chart at 1.13.1.
I am having VERY frequently tasks stuck in queued state, that don't get scheduled.

On the first task execution, on the frontend the task is stuck in 'queued', while on the scheduler:

[2024-05-14T08:53:11.792+0000] {scheduler_job_runner.py:639} INFO - Sending TaskInstanceKey(dag_id='etl-mongodb-postgres', task_id='count_before_pgraph', run_id='dataset_triggered__2024-05-14T08:53:10.825798+00:00', try_number=1, map_index=-1) to executor with priority 3 and queue default
[2024-05-14T08:53:11.792+0000] {base_executor.py:149} INFO - Adding to queue: ['airflow', 'tasks', 'run', 'etl-mongodb-postgres', 'count_before_pgraph', 'dataset_triggered__2024-05-14T08:53:10.825798+00:00', '--local', '--subdir', 'DAGS_FOLDER/lims_etl.py']
[2024-05-14T08:53:11.803+0000] {local_executor.py:90} INFO - QueuedLocalWorker running ['airflow', 'tasks', 'run', 'etl-mongodb-postgres', 'count_before_pgraph', 'dataset_triggered__2024-05-14T08:53:10.825798+00:00', '--local', '--subdir', 'DAGS_FOLDER/lims_etl.py']
...
[2024-05-14T08:53:12.232+0000] {task_command.py:426} INFO - Running <TaskInstance: etl-mongodb-postgres.count_before_pgraph dataset_triggered__2024-05-14T08:53:10.825798+00:00 [queued]> on host airflow-scheduler-779c5fbd96-458j7

When i try clearing it manually, logs show similarly a queued but still running; attempt=X and in the UI this task goes directly back to being shown as queued.

[2024-05-14T09:36:40.107+0000] {scheduler_job_runner.py:1595} INFO - Adopting or resetting orphaned tasks for active dag runs
[2024-05-14T09:36:41.910+0000] {scheduler_job_runner.py:417} INFO - 1 tasks up for execution:
        <TaskInstance: etl-mongodb-postgres.count_before_pgraph dataset_triggered__2024-05-14T08:53:10.825798+00:00 [scheduled]>
[2024-05-14T09:36:41.910+0000] {scheduler_job_runner.py:480} INFO - DAG etl-mongodb-postgres has 0/16 running and queued tasks
[2024-05-14T09:36:41.911+0000] {scheduler_job_runner.py:596} INFO - Setting the following tasks to queued state:
        <TaskInstance: etl-mongodb-postgres.count_before_pgraph dataset_triggered__2024-05-14T08:53:10.825798+00:00 [scheduled]>
[2024-05-14T09:36:41.914+0000] {scheduler_job_runner.py:639} INFO - Sending TaskInstanceKey(dag_id='etl-mongodb-postgres', task_id='count_before_pgraph', run_id='dataset_triggered__2024-05-14T08:53:10.825798+00:00', try_number=1, map_index=-1) to executor with priority 3 and queue default
[2024-05-14T09:36:41.914+0000] {base_executor.py:149} INFO - Adding to queue: ['airflow', 'tasks', 'run', 'etl-mongodb-postgres', 'count_before_pgraph', 'dataset_triggered__2024-05-14T08:53:10.825798+00:00', '--local', '--subdir', 'DAGS_FOLDER/lims_etl.py']
[2024-05-14T09:36:41.920+0000] {base_executor.py:284} INFO - queued but still running; attempt=1 task=TaskInstanceKey(dag_id='etl-mongodb-postgres', task_id='count_before_pgraph', run_id='dataset_triggered__2024-05-14T08:53:10.825798+00:00', try_number=1, map_index=-1)
[2024-05-14T09:36:42.080+0000] {base_executor.py:284} INFO - queued but still running; attempt=2 task=TaskInstanceKey(dag_id='etl-mongodb-postgres', task_id='count_before_pgraph', run_id='dataset_triggered__2024-05-14T08:53:10.825798+00:00', try_number=1, map_index=-1)
[2024-05-14T09:36:43.264+0000] {base_executor.py:284} INFO - queued but still running; attempt=3 task=TaskInstanceKey(dag_id='etl-mongodb-postgres', task_id='count_before_pgraph', run_id='dataset_triggered__2024-05-14T08:53:10.825798+00:00', try_number=1, map_index=-1)
[2024-05-14T09:36:44.402+0000] {base_executor.py:284} INFO - queued but still running; attempt=4 task=TaskInstanceKey(dag_id='etl-mongodb-postgres', task_id='count_before_pgraph', run_id='dataset_triggered__2024-05-14T08:53:10.825798+00:00', try_number=1, map_index=-1)
[2024-05-14T09:36:45.558+0000] {base_executor.py:284} INFO - queued but still running; attempt=5 task=TaskInstanceKey(dag_id='etl-mongodb-postgres', task_id='count_before_pgraph', run_id='dataset_triggered__2024-05-14T08:53:10.825798+00:00', try_number=1, map_index=-1)
[2024-05-14T09:36:45.923+0000] {base_executor.py:284} INFO - queued but still running; attempt=6 task=TaskInstanceKey(dag_id='etl-mongodb-postgres', task_id='count_before_pgraph', run_id='dataset_triggered__2024-05-14T08:53:10.825798+00:00', try_number=1, map_index=-1)
[2024-05-14T09:36:47.117+0000] {base_executor.py:284} INFO - queued but still running; attempt=7 task=TaskInstanceKey(dag_id='etl-mongodb-postgres', task_id='count_before_pgraph', run_id='dataset_triggered__2024-05-14T08:53:10.825798+00:00', try_number=1, map_index=-1)
[2024-05-14T09:36:48.347+0000] {base_executor.py:284} INFO - queued but still running; attempt=8 task=TaskInstanceKey(dag_id='etl-mongodb-postgres', task_id='count_before_pgraph', run_id='dataset_triggered__2024-05-14T08:53:10.825798+00:00', try_number=1, map_index=-1)
[2024-05-14T09:36:49.516+0000] {base_executor.py:284} INFO - queued but still running; attempt=9 task=TaskInstanceKey(dag_id='etl-mongodb-postgres', task_id='count_before_pgraph', run_id='dataset_triggered__2024-05-14T08:53:10.825798+00:00', try_number=1, map_index=-1)
[2024-05-14T09:36:50.677+0000] {base_executor.py:284} INFO - queued but still running; attempt=10 task=TaskInstanceKey(dag_id='etl-mongodb-postgres', task_id='count_before_pgraph', run_id='dataset_triggered__2024-05-14T08:53:10.825798+00:00', try_number=1, map_index=-1)
[2024-05-14T09:36:51.125+0000] {base_executor.py:284} INFO - queued but still running; attempt=11 task=TaskInstanceKey(dag_id='etl-mongodb-postgres', task_id='count_before_pgraph', run_id='dataset_triggered__2024-05-14T08:53:10.825798+00:00', try_number=1, map_index=-1)
[2024-05-14T09:36:52.311+0000] {base_executor.py:284} INFO - queued but still running; attempt=12 task=TaskInstanceKey(dag_id='etl-mongodb-postgres', task_id='count_before_pgraph', run_id='dataset_triggered__2024-05-14T08:53:10.825798+00:00', try_number=1, map_index=-1)
[2024-05-14T09:36:53.504+0000] {base_executor.py:287} ERROR - could not queue task TaskInstanceKey(dag_id='etl-mongodb-postgres', task_id='count_before_pgraph', run_id='dataset_triggered__2024-05-14T08:53:10.825798+00:00', try_number=1, map_index=-1) (still running after 12 attempts)

Much later, I saw the follow up task (refresh_pipeline_graph) being scheduled, upon which all of a sudden the previously stuck task (count_before_pgraph) signals Received executor event with state success.

[2024-05-14T10:44:40.831+0000] {scheduler_job_runner.py:417} INFO - 1 tasks up for execution:
        <TaskInstance: etl-mongodb-postgres.refresh_pipeline_graph dataset_triggered__2024-05-14T08:53:10.825798+00:00 [scheduled]>
[2024-05-14T10:44:40.831+0000] {scheduler_job_runner.py:480} INFO - DAG etl-mongodb-postgres has 0/16 running and queued tasks
[2024-05-14T10:44:40.832+0000] {scheduler_job_runner.py:596} INFO - Setting the following tasks to queued state:
        <TaskInstance: etl-mongodb-postgres.refresh_pipeline_graph dataset_triggered__2024-05-14T08:53:10.825798+00:00 [scheduled]>
[2024-05-14T10:44:40.836+0000] {scheduler_job_runner.py:639} INFO - Sending TaskInstanceKey(dag_id='etl-mongodb-postgres', task_id='refresh_pipeline_graph', run_id='dataset_triggered__2024-05-14T08:53:10.825798+00:00', try_number=1, map_index=-1) to executor with priority 2 and queue default
[2024-05-14T10:44:40.836+0000] {base_executor.py:149} INFO - Adding to queue: ['airflow', 'tasks', 'run', 'etl-mongodb-postgres', 'refresh_pipeline_graph', 'dataset_triggered__2024-05-14T08:53:10.825798+00:00', '--local', '--subdir', 'DAGS_FOLDER/lims_etl.py']
[2024-05-14T10:44:40.842+0000] {scheduler_job_runner.py:689} INFO - Received executor event with state success for task instance TaskInstanceKey(dag_id='etl-mongodb-postgres', task_id='count_before_pgraph', run_id='dataset_triggered__2024-05-14T08:53:10.825798+00:00', try_number=1, map_index=-1)

So, somehow it looks like the task started running at 2024-05-14T08:53:11, somehow this wasn't recorded, and it was running until 2024-05-14T10:44:40. This task would take around 1s to finish usually.