**Apache Airflow version**: 2.0.2

**Kubernetes version (if you are using kubernetes)** (use `kubectl version`): Running on Docker 19.03.13

**Environment**:

- **Cloud provider or hardware configuration**:
- **OS** (e.g. from /etc/os-release): Windows 10 Enterprise
- **Kernel**:
- **Install tools**:
- **Others**:

**What happened**:

I receive an HTTP Error 500 when changing tasks status through the REST API.


**What you expected to happen**:

I expected to receive a HTTP 200.

**How to reproduce it**:

First, we trigger a new Dag Run:
```
    dag_id = 'test'
    run_id = 1000

    r = requests.post('http://localhost:8080/api/v1/dags/' + dag_id + '/dagRuns', 
        json={"dag_run_id": str(run_id), "conf": { } },
        auth=HTTPBasicAuth('airflow', 'airflow'))

    if r.status_code == 200:
        print("Dag started with run_id", run_id)
```

Then we try to abort the DAG Run:
```
    r = requests.get('http://localhost:8080/api/v1/dags/' + dag_id + '/dagRuns/' + str(run_id) + '/taskInstances?state=running', 
        auth=HTTPBasicAuth('airflow', 'airflow'))

    task_id = r.json()['task_instances'][0]['task_id']
    execution_date = r.json()['task_instances'][0]['execution_date']

    r = requests.post('http://localhost:8080/api/v1/dags/' + dag_id + '/updateTaskInstancesState', 
        json={"task_id": str(task_id), 
            "execution_date": str(execution_date),
            "include_upstream": True,
            "include_downstream": True,
            "include_future": True,
            "include_past": False,
            "new_state": "failed" 
        },
        auth=HTTPBasicAuth('airflow', 'airflow'))

    print(r.status_code)
```

**Anything else we need to know**:

This is the server side track:
```
Something bad has happened.
Please consider letting us know by creating a <b><a href="https://github.com/apache/airflow/issues/new/choose">bug report using GitHub</a></b>.

Python version: 3.6.13
Airflow version: 2.0.2
Node: c8d75444cd4a
-------------------------------------------------------------------------------
Traceback (most recent call last):
  File &#34;/home/airflow/.local/lib/python3.6/site-packages/flask/app.py&#34;, line 2447, in wsgi_app
    response = self.full_dispatch_request()
  File &#34;/home/airflow/.local/lib/python3.6/site-packages/flask/app.py&#34;, line 1952, in full_dispatch_request
    rv = self.handle_user_exception(e)
  File &#34;/home/airflow/.local/lib/python3.6/site-packages/flask/app.py&#34;, line 1821, in handle_user_exception
    reraise(exc_type, exc_value, tb)
  File &#34;/home/airflow/.local/lib/python3.6/site-packages/flask/_compat.py&#34;, line 39, in reraise
    raise value
  File &#34;/home/airflow/.local/lib/python3.6/site-packages/flask/app.py&#34;, line 1950, in full_dispatch_request
    rv = self.dispatch_request()
  File &#34;/home/airflow/.local/lib/python3.6/site-packages/flask/app.py&#34;, line 1936, in dispatch_request
    return self.view_functions[rule.endpoint](**req.view_args)
  File &#34;/home/airflow/.local/lib/python3.6/site-packages/connexion/decorators/decorator.py&#34;, line 48, in wrapper
    response = function(request)
  File &#34;/home/airflow/.local/lib/python3.6/site-packages/connexion/decorators/uri_parsing.py&#34;, line 144, in wrapper
    response = function(request)
  File &#34;/home/airflow/.local/lib/python3.6/site-packages/connexion/decorators/validation.py&#34;, line 184, in wrapper
    response = function(request)
  File &#34;/home/airflow/.local/lib/python3.6/site-packages/connexion/decorators/validation.py&#34;, line 384, in wrapper
    return function(request)
  File &#34;/home/airflow/.local/lib/python3.6/site-packages/connexion/decorators/response.py&#34;, line 103, in wrapper
    response = function(request)
  File &#34;/home/airflow/.local/lib/python3.6/site-packages/connexion/decorators/parameter.py&#34;, line 121, in wrapper
    return function(**kwargs)
  File &#34;/home/airflow/.local/lib/python3.6/site-packages/airflow/api_connexion/security.py&#34;, line 47, in decorated
    return func(*args, **kwargs)
  File &#34;/home/airflow/.local/lib/python3.6/site-packages/airflow/utils/session.py&#34;, line 70, in wrapper
    return func(*args, session=session, **kwargs)
  File &#34;/home/airflow/.local/lib/python3.6/site-packages/airflow/api_connexion/endpoints/task_instance_endpoint.py&#34;, line 314, in post_set_task_instances_state
    commit=not data[&#34;dry_run&#34;],
KeyError: &#39;dry_run&#39;
```



With every call.

tests/api_connexion/endpoints/test_task_instance_endpoint.py
]
        }
        dag = self.app.dag_bag.dags['example_python_operator']  
        task = dag.task_dict['print_the_context']
        mock_set_state.assert_called_once_with(
            commit=False,
            downstream=True,
            execution_date=DEFAULT_DATETIME_1,
            future=True,
            past=True,
            state='failed',
            tasks=[task],
            upstream=True,
        )
    def test_should_raises_401_unauthenticated(self):
        response = self.client.post(
            "/api/v1/dags/example_python_operator/updateTaskInstancesState",
            json={
                "dry_run": True,
                "task_id": "print_the_context",
                "execution_date": DEFAULT_DATETIME_1,
                "include_upstream": True,
                "include_downstream": True,
                "include_future": True,
                "include_past": True,
                "new_state": "failed",
            },
        )
        assert_401(response)
    def test_should_raise_403_forbidden(self):