The executor reported that the task instance <TaskInstance: ... [queued]> finished with state success

[antonysouthworth-halter]

Yes, quite similar to a few other discussions:

• Tasks being marked as failed even after running successfully #34413
• Executor reports task instance <TaskInstance: XXX [queued]> finished (failed) although the task says it's queued. (Info: None) Was the task killed externally #42966
• backfill #42852

I am also observing tasks being marked as SUCCESS then immediately marked as FAILED. Something that I believe to be different in this case compared to the others linked is my tasks are completing very quickly (<1s) which I wonder if that's contributing?

Specifically in this case I a have a DAG that builds some data since 2024-09-11, and I am using airflow dags backfill CLI to backfill this DAG between 2022-01-01 and 2024-09-10.

What I am observing is a number of the backfill DagRuns failing all with the same behaviour; event log shows task marked success and then failed very soon after, and the infamous log line mentioned in the title.

Here's a snippet of collected logs: lines starting with task are logs from the Airflow worker executing the tasks, lines starting with backf are logs from the Airflow CLI:

task 2024-10-14T02:19:10.388Z Resuming after deferral
task 2024-10-14T02:19:11.013Z Marking task as SUCCESS.
task 2024-10-14T02:19:11.156Z Task exited with return code 0
task 2024-10-14T02:19:11.217Z 1 downstream tasks scheduled from follow-on schedule check
bakf 2024-10-14T02:19:11.826Z {backfill_job_runner.py:310} DEBUG - Executor state: success task <TaskInstance: ... [queued]>
bakf 2024-10-14T02:19:11.828Z {taskinstance.py:3303} ERROR - The executor reported that the task instance <TaskInstance: ... [queued]> finished with state success, but the task instance's state attribute is queued. Learn more: https://airflow.apache.org/docs/apache-airflow/stable/troubleshooting.html#task-state-changed-externally

The task simply starts an Athena query using AthenaOperator, deferred polling until it completes. So when the deferral completes the task finishes almost immediately; I do wonder if there is some kind of race condition between scheduler updating from QUEUED->RUNNING and executor updating from RUNNING->SUCCESS.

My cluster right now only has 5x DAGs total, each with 1-3 tasks each. So a very small deployment (1x Fargate task each for scheduler, worker, web, triggerer, flower, db.t4g.medium RDS postgres for metadb, cache.t2.micro for broker) and I don't believe I am hitting any of the timeouts mentioned in other threads (Celery operation timeout, dagbag import/process, etc.).

[antonysouthworth-halter]

You can see it sometimes happen for the first operator too; it's an "allowed lateness" operator (#42636) which also succeeds very quickly as the lateness cutoff has already elapsed (i.e. this DateTimeSensorAsync is effectively trying to wait for zero time in this instance).

[antonysouthworth-halter]

Weirdly I do not observe this in scheduled__ runs, only seems to be affecting backfill__ runs.

[antonysouthworth-halter]

Also forgot to mention we are on Airflow 2.10.0.

[antonysouthworth-halter]

I wonder if https://cwiki.apache.org/confluence/display/AIRFLOW/AIP-78+Scheduler-managed+backfill will fix? Potentially a bug in the backfill CLI scheduler that does not exist in the "real" scheduler? (Based on my observation that this seems to only happen for my backfill jobs).

[antonysouthworth-halter]

I found two copies of the same Airflow task in Celery Flower UI with different UUIDS, so looks like I am somehow running the task twice.

First one:

Second one:

[antonysouthworth-halter]

[antonysouthworth-halter]

EDIT nvm this is just the two tasks in the DAG running one after the other.

[antonysouthworth-halter]

event log

[antonysouthworth-halter]

Suspicious log

{backfill_job_runner.py:303} WARNING - TaskInstanceKey(dag_id='service_daily', task_id='devices_daily_snapshot_healthck', run_id='backfill__2024-09-04T00:00:00+00:00', try_number=5, map_index=-1) state queued not in running=dict_values([<TaskInstance: service_daily.devices_daily_snapshot backfill__2024-09-04T00:00:00+00:00 [queued]>, <TaskInstance: service_daily.devices_daily_snapshot_healthck backfill__2024-09-04T00:00:00+00:00 [queued]>])

[antonysouthworth-halter]

Ahhh right, the reason for the second Celery message is because we are using deferred operator. First message is the task starting, second message is task resuming after trigger event.

[antonysouthworth-halter]

Ok! Testing out #43063 on my cluster I am seeing the log line I added get exercised and it is not followed by the infamous The executor reported that the task instance finished with state success, but the task instance's state attribute is queued message.

[2024-10-17T01:27:20.695+0000] {backfill_job_runner.py:199} DEBUG - Task instance <TaskInstance: service_daily.allowed_lateness backfill__2022-01-17T00:00:00+00:00 [success]> succeeded. Don't rerun.
[2024-10-17T01:27:20.686+0000] {backfill_job_runner.py:316} DEBUG - In-memory TaskInstance state <TaskInstance: service_daily.allowed_lateness backfill__2022-01-17T00:00:00+00:00 [queued]> does not agree with executor state success. Attempting to resolve by refreshing in-memory task instance from DB.
[2024-10-17T01:27:20.685+0000] {backfill_job_runner.py:310} DEBUG - Executor state: success task <TaskInstance: service_daily.allowed_lateness backfill__2022-01-17T00:00:00+00:00 [queued]>