### Apache Airflow version

2.2.0b2 (beta snapshot)

### Operating System

debian (docker)

### Versions of Apache Airflow Providers

n/a

### Deployment

Astronomer

### Deployment details

`astro dev start` with dockerfile:

```
FROM quay.io/astronomer/ap-airflow-dev:2.2.0-buster-44011
```

And dags:
- [simple_branch](https://gist.github.com/MatrixManAtYrService/bb7eac11d2a60c840b02f5fc7c4e864f#file-simple_branch-py)
- [branch](https://gist.github.com/MatrixManAtYrService/bb7eac11d2a60c840b02f5fc7c4e864f#file-branch-py)

### What happened

Spamming the "manual trigger" button on `simple_branch` caused no trouble:

![good](https://user-images.githubusercontent.com/5834582/134444526-4c733e74-3a86-4043-8aa3-a5475c93cc0b.png)

But doing the same on `branch` caused intermittent failures:

![bad](https://user-images.githubusercontent.com/5834582/134444555-161b1803-0e78-41f8-a1d3-25e652001fd1.png)

task logs:
```
*** Reading local file: /usr/local/airflow/logs/branch/decide_if_you_like_it/2021-09-23T01:31:50.261287+00:00/2.log
[2021-09-23, 01:41:27 UTC] {taskinstance.py:1034} INFO - Dependencies all met for <TaskInstance: branch.decide_if_you_like_it manual__2021-09-23T01:31:50.261287+00:00 [queued]>
[2021-09-23, 01:41:27 UTC] {taskinstance.py:1034} INFO - Dependencies all met for <TaskInstance: branch.decide_if_you_like_it manual__2021-09-23T01:31:50.261287+00:00 [queued]>
[2021-09-23, 01:41:27 UTC] {taskinstance.py:1232} INFO - 
--------------------------------------------------------------------------------
[2021-09-23, 01:41:27 UTC] {taskinstance.py:1233} INFO - Starting attempt 2 of 2
[2021-09-23, 01:41:27 UTC] {taskinstance.py:1234} INFO - 
--------------------------------------------------------------------------------
[2021-09-23, 01:41:27 UTC] {taskinstance.py:1253} INFO - Executing <Task(BranchPythonOperator): decide_if_you_like_it> on 2021-09-23 01:31:50.261287+00:00
[2021-09-23, 01:41:27 UTC] {standard_task_runner.py:52} INFO - Started process 8584 to run task
[2021-09-23, 01:41:27 UTC] {standard_task_runner.py:76} INFO - Running: ['airflow', 'tasks', 'run', 'branch', 'decide_if_you_like_it', 'manual__2021-09-23T01:31:50.261287+00:00', '--job-id', '216', '--raw', '--subdir', 'DAGS_FOLDER/branch.py', '--cfg-path', '/tmp/tmp82f81d8w', '--error-file', '/tmp/tmpco74n1v0']
[2021-09-23, 01:41:27 UTC] {standard_task_runner.py:77} INFO - Job 216: Subtask decide_if_you_like_it
[2021-09-23, 01:41:27 UTC] {logging_mixin.py:109} INFO - Running <TaskInstance: branch.decide_if_you_like_it manual__2021-09-23T01:31:50.261287+00:00 [running]> on host 992912b473f9
[2021-09-23, 01:41:27 UTC] {taskinstance.py:1406} INFO - Exporting the following env vars:
AIRFLOW_CTX_DAG_OWNER=airflow
AIRFLOW_CTX_DAG_ID=branch
AIRFLOW_CTX_TASK_ID=decide_if_you_like_it
AIRFLOW_CTX_EXECUTION_DATE=2021-09-23T01:31:50.261287+00:00
AIRFLOW_CTX_DAG_RUN_ID=manual__2021-09-23T01:31:50.261287+00:00
[2021-09-23, 01:41:27 UTC] {python.py:152} INFO - Done. Returned value was: None
[2021-09-23, 01:41:27 UTC] {skipmixin.py:143} INFO - Following branch None
[2021-09-23, 01:41:27 UTC] {taskinstance.py:1680} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1315, in _run_raw_task
    self._prepare_and_execute_task_with_callbacks(context, task)
  File "/usr/local/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1437, in _prepare_and_execute_task_with_callbacks
    result = self._execute_task(context, task_copy)
  File "/usr/local/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1493, in _execute_task
    result = execute_callable(context=context)
  File "/usr/local/lib/python3.9/site-packages/airflow/operators/python.py", line 181, in execute
    self.skip_all_except(context['ti'], branch)
  File "/usr/local/lib/python3.9/site-packages/airflow/models/skipmixin.py", line 147, in skip_all_except
    branch_task_ids = set(branch_task_ids)
TypeError: 'NoneType' object is not iterable
[2021-09-23, 01:41:27 UTC] {taskinstance.py:1261} INFO - Marking task as FAILED. dag_id=branch, task_id=decide_if_you_like_it, execution_date=20210923T013150, start_date=20210923T014127, end_date=20210923T014127
[2021-09-23, 01:41:27 UTC] {standard_task_runner.py:88} ERROR - Failed to execute job 216 for task decide_if_you_like_it
Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/airflow/task/task_runner/standard_task_runner.py", line 85, in _start_by_fork
    args.func(args, dag=self.dag)
  File "/usr/local/lib/python3.9/site-packages/airflow/cli/cli_parser.py", line 48, in command
    return func(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/airflow/utils/cli.py", line 92, in wrapper
    return f(*args, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/airflow/cli/commands/task_command.py", line 292, in task_run
    _run_task_by_selected_method(args, dag, ti)
  File "/usr/local/lib/python3.9/site-packages/airflow/cli/commands/task_command.py", line 107, in _run_task_by_selected_method
    _run_raw_task(args, ti)
  File "/usr/local/lib/python3.9/site-packages/airflow/cli/commands/task_command.py", line 180, in _run_raw_task
    ti._run_raw_task(
  File "/usr/local/lib/python3.9/site-packages/airflow/utils/session.py", line 70, in wrapper
    return func(*args, session=session, **kwargs)
  File "/usr/local/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1315, in _run_raw_task
    self._prepare_and_execute_task_with_callbacks(context, task)
  File "/usr/local/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1437, in _prepare_and_execute_task_with_callbacks
    result = self._execute_task(context, task_copy)
  File "/usr/local/lib/python3.9/site-packages/airflow/models/taskinstance.py", line 1493, in _execute_task
    result = execute_callable(context=context)
  File "/usr/local/lib/python3.9/site-packages/airflow/operators/python.py", line 181, in execute
    self.skip_all_except(context['ti'], branch)
  File "/usr/local/lib/python3.9/site-packages/airflow/models/skipmixin.py", line 147, in skip_all_except
    branch_task_ids = set(branch_task_ids)
TypeError: 'NoneType' object is not iterable
[2021-09-23, 01:41:27 UTC] {local_task_job.py:154} INFO - Task exited with return code 1
[2021-09-23, 01:41:27 UTC] {local_task_job.py:264} INFO - 0 downstream tasks scheduled from follow-on schedule check
```


### What you expected to happen

`BranchPythonOperator` should create tasks that always succeed

### How to reproduce

Keep clicking manual executions of the dag called `branch` until you've triggered ten or so.  At least one of them will fail with the error:

```
TypeError: 'NoneType' object is not iterable
```

### Anything else

I've also seen this behavior in image: `quay.io/astronomer/ap-airflow-dev:2.1.4-buster-44000`

### Are you willing to submit PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [X] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

tests/operators/test_python.py
def __init__(self, *args, **kwargs):
        self.args = args
        self.kwargs = kwargs
def build_recording_function(calls_collection):
    """
    We can not use a Mock instance as a PythonOperator callable function or some tests fail with a
    TypeError: Object of type Mock is not JSON serializable
    Then using this custom function recording custom Call objects for further testing
    (replacing Mock.assert_called_with assertion method)
    """
    def recording_function(*args, **kwargs):
        calls_collection.append(Call(*args, **kwargs))
    return recording_function
class TestPythonBase(unittest.TestCase):
    """Base test class for TestPythonOperator and TestPythonSensor classes"""
    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        with create_session() as session:
            session.query(DagRun).delete()
            session.query(TI).delete()
    def setUp(self):
        super().setUp()
        self.dag = DAG('test_dag', default_args={'owner': 'airflow', 'start_date': DEFAULT_DATE})
        self.addCleanup(self.dag.clear)
        self.clear_run()
        self.addCleanup(self.clear_run)
    def tearDown(self):