### 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)

airflow/operators/python.py
context.update(self.op_kwargs)
        context['templates_dict'] = self.templates_dict
        self.op_kwargs = determine_kwargs(self.python_callable, self.op_args, context)
        return_value = self.execute_callable()
        self.log.info("Done. Returned value was: %s", return_value)
        return return_value
    def execute_callable(self):
        """
        Calls the python callable with the given arguments.
        :return: the return value of the call.
        :rtype: any
        """
        return self.python_callable(*self.op_args, **self.op_kwargs)
class BranchPythonOperator(PythonOperator, SkipMixin):
    """
    Allows a workflow to "branch" or follow a path following the execution
    of this task.
    It derives the PythonOperator and expects a Python function that returns
    a single task_id or list of task_ids to follow. The task_id(s) returned
    should point to a task directly downstream from {self}. All other "branches"
    or directly downstream tasks are marked with a state of ``skipped`` so that
    these paths can't move forward. The ``skipped`` states are propagated
    downstream to allow for the DAG state to fill up and the DAG run's state
    to be inferred.
    """
    def execute(self, context: Dict):
        branch = super().execute(context)
        self.skip_all_except(context['ti'], branch)
        return branch
class ShortCircuitOperator(PythonOperator, SkipMixin):