Airflow stable API taskInstance call fails if a task is removed from running DAG

[soltanianalytics]

Apache Airflow version: 2.0.1

Environment: Docker on Win 10 with WSL, image based on apache/airflow:2.0.1-python3.8

What happened:

I'm using the airflow API and the following (what I believe to be a) bug popped up:

>>> import requests
>>> r = requests.get("http://localhost:8084/api/v1/dags/~/dagRuns/~/taskInstances", auth=HTTPBasicAuth('username', 'password'))
>>> r.status_code
500
>>> print(r.text)
{
  "detail": "'removed' is not one of ['success', 'running', 'failed', 'upstream_failed', 'skipped', 'up_for_retry', 'up_for_reschedule', 'queued', 'none', 'scheduled']\n\nFailed validating 'enum' in schema['allOf'][0]['properties']['task_instances']['items']['properties']['state']:\n    {'description': 'Task state.',\n     'enum': ['success',\n              'running',\n              'failed',\n              'upstream_failed',\n              'skipped',\n              'up_for_retry',\n              'up_for_reschedule',\n              'queued',\n              'none',\n              'scheduled'],\n     'nullable': True,\n     'type': 'string',\n     'x-scope': ['',\n                 '#/components/schemas/TaskInstanceCollection',\n                 '#/components/schemas/TaskInstance']}\n\nOn instance['task_instances'][16]['state']:\n    'removed'",
  "status": 500,
  "title": "Response body does not conform to specification",
  "type": "https://airflow.apache.org/docs/2.0.1rc2/stable-rest-api-ref.html#section/Errors/Unknown"
}
>>> print(r.json()["detail"])
'removed' is not one of ['success', 'running', 'failed', 'upstream_failed', 'skipped', 'up_for_retry', 'up_for_reschedule', 'queued', 'none', 'scheduled']

Failed validating 'enum' in schema['allOf'][0]['properties']['task_instances']['items']['properties']['state']:
    {'description': 'Task state.',
     'enum': ['success',
              'running',
              'failed',
              'upstream_failed',
              'skipped',
              'up_for_retry',
              'up_for_reschedule',
              'queued',
              'none',
              'scheduled'],
     'nullable': True,
     'type': 'string',
     'x-scope': ['',
                 '#/components/schemas/TaskInstanceCollection',
                 '#/components/schemas/TaskInstance']}

On instance['task_instances'][16]['state']:
    'removed'

This happened after I changed a DAG in the corresponding instance, thus a task was removed from a DAG while the DAG was running.

What you expected to happen:

Give me all task instances, whether including the removed ones or not is up to the airflow team to decide (no preferences from my side, though I'd guess it makes more sense to supply all data as it is available).

How to reproduce it:

• Run airflow
• Create a DAG with multiple tasks
• While the DAG is running, remove one of the tasks (ideally one that did not yet run)
• Make the API call as above

[ephraimbuddy]

From here:

airflow/airflow/utils/state.py

Lines 48 to 60 in a7e4266

	task_states = (
	SUCCESS,
	RUNNING,
	FAILED,
	UPSTREAM_FAILED,
	SKIPPED,
	UP_FOR_RETRY,
	UP_FOR_RESCHEDULE,
	QUEUED,
	NONE,
	SCHEDULED,
	SENSING,
	)

The task states do not include removed, I think that's why it was omitted here:

airflow/airflow/api_connexion/openapi/v1.yaml

Lines 2500 to 2509 in a7e4266

	- success
	- running
	- failed
	- upstream_failed
	- skipped
	- up_for_retry
	- up_for_reschedule
	- queued
	- none
	- scheduled

Should we change this @kaxil?

[kaxil]

We just need to check it (`task_states) isn't used anywhere else to cause domino effect