**Apache Airflow version**:
2.1.2


**Environment**:

- **Cloud provider or hardware configuration**: AWS ECS
- **OS** (e.g. from /etc/os-release): debian
- **Kernel** (e.g. `uname -a`): Linux
- **Install tools**:
- **Others**:

**What happened**:
When setting default_impersonation in airflow.cfg as `testuser` and not passing run_as_user in the default_parm or operator parameters, it is showing run_as_user in the UI as None (It is using testuser to execute the dag though):
![image](https://user-images.githubusercontent.com/25225354/126956597-089b29db-f833-4620-92fa-d26f7ad29b6f.png)
And also throwing error in logs and some of the dags are failing
```
[2021-07-26 09:53:41,929: ERROR/ForkPoolWorker-7] Failed to execute task PID of job runner does not match.
--
Traceback (most recent call last):
File "/usr/local/lib/python3.7/site-packages/airflow/executors/celery_executor.py", line 117, in _execute_in_fork
args.func(args)
File "/usr/local/lib/python3.7/site-packages/airflow/cli/cli_parser.py", line 48, in command
return func(*args, **kwargs)
File "/usr/local/lib/python3.7/site-packages/airflow/utils/cli.py", line 91, in wrapper
return f(*args, **kwargs)
File "/usr/local/lib/python3.7/site-packages/airflow/cli/commands/task_command.py", line 238, in task_run
_run_task_by_selected_method(args, dag, ti)
File "/usr/local/lib/python3.7/site-packages/airflow/cli/commands/task_command.py", line 64, in _run_task_by_selected_method
_run_task_by_local_task_job(args, ti)
File "/usr/local/lib/python3.7/site-packages/airflow/cli/commands/task_command.py", line 121, in _run_task_by_local_task_job
run_job.run()
File "/usr/local/lib/python3.7/site-packages/airflow/jobs/base_job.py", line 245, in run
self._execute()
File "/usr/local/lib/python3.7/site-packages/airflow/jobs/local_task_job.py", line 131, in _execute
self.heartbeat()
File "/usr/local/lib/python3.7/site-packages/airflow/jobs/base_job.py", line 226, in heartbeat
self.heartbeat_callback(session=session)
File "/usr/local/lib/python3.7/site-packages/airflow/utils/session.py", line 67, in wrapper
return func(*args, **kwargs)
File "/usr/local/lib/python3.7/site-packages/airflow/jobs/local_task_job.py", line 195, in heartbeat_callback
raise AirflowException("PID of job runner does not match")
airflow.exceptions.AirflowException: PID of job runner does not match
```

**What you expected to happen**:
Dags not to fail and logs not to show this error messages.

**How to reproduce it**:
Set default_impersonation in airflow.cfg and don't pass run_as_user in task
Run a dag with multiple task (more than 1) which runs for more than 10sec 

**Anything else we need to know**:
I suspect that the if-else statement [here](https://github.com/apache/airflow/blob/main/airflow/jobs/local_task_job.py#L196) is causing the issue

Also I am trying to set run_as_user for all dags to be `testuser` using cluster policy, doing that also its giving the same error
```
def task_policy(task):
  task.run_as_user = 'testuser'
```

airflow/jobs/local_task_job.py
max_wait_time = max(
                    0,  
                    min(
                        (
                            heartbeat_time_limit
                            - (timezone.utcnow() - self.latest_heartbeat).total_seconds() * 0.75
                        ),
                        self.heartrate,
                    ),
                )
                return_code = self.task_runner.return_code(timeout=max_wait_time)
                if return_code is not None:
                    self.handle_task_exit(return_code)
                    return
                self.heartbeat()
                
                
                
                time_since_last_heartbeat = (timezone.utcnow() - self.latest_heartbeat).total_seconds()
                if time_since_last_heartbeat > heartbeat_time_limit:
                    Stats.incr('local_task_job_prolonged_heartbeat_failure', 1, 1)
                    self.log.error("Heartbeat time limit exceeded!")
                    raise AirflowException(
                        "Time since last heartbeat({:.2f}s) "
                        "exceeded limit ({}s).".format(time_since_last_heartbeat, heartbeat_time_limit)
                    )
        finally:
            self.on_kill()
    def handle_task_exit(self, return_code: int) -> None: