### 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
#
#
#
import inspect
import os
import pickle
import shutil
import sys
import types
import warnings
from tempfile import TemporaryDirectory
from textwrap import dedent
from typing import Any, Callable, Dict, Iterable, List, Optional, Union
import dill
from airflow.exceptions import AirflowException
from airflow.models import BaseOperator
from airflow.models.skipmixin import SkipMixin
from airflow.models.taskinstance import _CURRENT_CONTEXT
from airflow.utils.operator_helpers import determine_kwargs
from airflow.utils.process_utils import execute_in_subprocess
from airflow.utils.python_virtualenv import prepare_virtualenv, write_python_script
def task(python_callable: Optional[Callable] = None, multiple_outputs: Optional[bool] = None, **kwargs):
    """
    Deprecated function that calls @task.python and allows users to turn a python function into
    an Airflow task. Please use the following instead: