Regression on pid reset to allow task start after heartbeat - #17333

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst)
Here are some useful points:

• Pay attention to the quality of your code (flake8, mypy and type annotations). Our pre-commits will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example DAG that shows how users should use it.
• Consider using Breeze environment for testing locally, it’s a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

Do you have a dag that could be used to reproduce this behaviour?

Do you have a dag that could be used to reproduce this behaviour?

@ephraimbuddy IMO it's not a bug that can be reproduced in a simply dag cause it not concern task execution code but task runner.
To reproduce we have to slowdown the task starting process:

• In the task pid update in https://github.com/apache/airflow/blob/main/airflow/models/taskinstance.py#L1180
• Or in the task fork from runner after forking https://github.com/apache/airflow/blob/main/airflow/task/task_runner/standard_task_runner.py#L54
• Or in the raw task command line (executed by scheduler) https://github.com/apache/airflow/blob/main/airflow/cli/commands/task_command.py#L156

If the starting process take more time than heartbeat (5s default), and it's not the first TI run, the problem occur.

I don't know if you have another idea to reproduce that in a dag ?

In my case, it occur more specifically in production at night, when the scheduler process our data of last day, and lot of task is running in parallel that generate load on our servers and our airflow db.

Maybe our first log with that error from airflow can help

[2021-07-18 03:09:34,879] {{logging_mixin.py:104}} INFO - Running <TaskInstance: task_name 2021-07-17T00:00:00+00:00 [running]> on host xxxxxxxxxxx
[2021-07-18 03:09:34,885] {{local_task_job.py:195}} WARNING - Recorded pid 4932 does not match the current pid 14922
[2021-07-18 03:09:34,893] {{process_utils.py:100}} INFO - Sending Signals.SIGTERM to GPID 14922
[2021-07-18 03:09:34,900] {{process_utils.py:66}} INFO - Process psutil.Process(pid=14922, status='terminated', exitcode=<Negsignal.SIGTERM: -15>, started='03:09:29') (14922) terminated with exit code Negsignal.SIGTERM

LGTM @ephraimbuddy

Awesome work, congrats on your first merged pull request!

I incorporated the change https://github.com/apache/airflow/blob/2.0.2/airflow/models/taskinstance.py#L1076 and I am still getting the error ....