Proper warning message when recorded PID is different from current PID (

#17411)

Currently, when the recorded PID is different from the current PID, in
the case of run_as_user, the warning is not clear because ti.pid is used
as the recorded PID instead of parent process of ti.pid. In this case,
users would see that the PIDs are the same but there was a warning that
they are not the same

This change fixes it.

(cherry picked from commit a4b6f1c)

airflow/jobs/local_task_job.py

@@ -187,14 +187,17 @@ def heartbeat_callback(self, session=None):
                 )
                 raise AirflowException("Hostname of job runner does not match")
             current_pid = self.task_runner.process.pid
-
-            same_process = ti.pid == current_pid
+            recorded_pid = ti.pid
+            same_process = recorded_pid == current_pid
 
             if ti.run_as_user or self.task_runner.run_as_user:
-                same_process = psutil.Process(ti.pid).ppid() == current_pid
+                recorded_pid = psutil.Process(ti.pid).ppid()
+                same_process = recorded_pid == current_pid
 
-            if ti.pid is not None and not same_process:
-                self.log.warning("Recorded pid %s does not match " "the current pid %s", ti.pid, current_pid)
+            if recorded_pid is not None and not same_process:
+                self.log.warning(
+                    "Recorded pid %s does not match the current pid %s", recorded_pid, current_pid
+                )
                 raise AirflowException("PID of job runner does not match")
         elif self.task_runner.return_code() is None and hasattr(self.task_runner, 'process'):
             self.log.warning(