[airflow] 39/39: Proper warning message when recorded PID is different from current PID (#17411)

[jh...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

jhtimmins pushed a commit to branch v2-1-test
in repository https://gitbox.apache.org/repos/asf/airflow.git

commit 0bd955831657a0ce628577983e5fb9ca16b7875c
Author: Ephraim Anierobi <sp...@gmail.com>
AuthorDate: Wed Aug 4 13:32:05 2021 +0100

    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 a4b6f1c1c5ab92fd5b623f119263d83bd46ab2e6)
---
 airflow/jobs/local_task_job.py | 13 ++++++++-----
 1 file changed, 8 insertions(+), 5 deletions(-)

diff --git a/airflow/jobs/local_task_job.py b/airflow/jobs/local_task_job.py
index af7b317..3afc801 100644
--- a/airflow/jobs/local_task_job.py
+++ b/airflow/jobs/local_task_job.py
@@ -187,14 +187,17 @@ class LocalTaskJob(BaseJob):
                 )
                 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(