[AIRFLOW-6730] Use total_seconds instead of seconds

airflow/jobs/base_job.py

@@ -118,7 +118,7 @@ def is_alive(self, grace_multiplier=2.1):
         """
         return (
             self.state == State.RUNNING and
-            (timezone.utcnow() - self.latest_heartbeat).seconds < self.heartrate * grace_multiplier
+            (timezone.utcnow() - self.latest_heartbeat).total_seconds() < self.heartrate * grace_multiplier
         )
 
     @provide_session

airflow/jobs/scheduler_job.py

@@ -941,7 +941,7 @@ def is_alive(self, grace_multiplier=None):
         scheduler_health_check_threshold = conf.getint('scheduler', 'scheduler_health_check_threshold')
         return (
             self.state == State.RUNNING and
-            (timezone.utcnow() - self.latest_heartbeat).seconds < scheduler_health_check_threshold
+            (timezone.utcnow() - self.latest_heartbeat).total_seconds() < scheduler_health_check_threshold
         )
 
     @provide_session

airflow/kubernetes/pod_launcher.py

@@ -111,7 +111,7 @@ def run_pod(
         if resp.status.start_time is None:
             while self.pod_not_started(pod):
                 delta = dt.now() - curr_time
-                if delta.seconds >= startup_timeout:
+                if delta.total_seconds() >= startup_timeout:
                     raise AirflowException("Pod took too long to start")
                 time.sleep(1)
             self.log.debug('Pod not yet started')

airflow/providers/google/cloud/operators/dataproc.py

@@ -244,7 +244,7 @@ def _get_init_action_timeout(self):
                 return self.init_action_timeout
             elif match.group(2) == "m":
                 val = float(match.group(1))
-                return "{}s".format(timedelta(minutes=val).seconds)
+                return "{}s".format(int(timedelta(minutes=val).total_seconds()))
 
         raise AirflowException(
             "DataprocClusterCreateOperator init_action_timeout"
@@ -621,13 +621,13 @@ def _graceful_decommission_timeout_object(self) -> Optional[Dict]:
                 timeout = int(match.group(1))
             elif match.group(2) == "m":
                 val = float(match.group(1))
-                timeout = timedelta(minutes=val).seconds
+                timeout = int(timedelta(minutes=val).total_seconds())
             elif match.group(2) == "h":
                 val = float(match.group(1))
-                timeout = timedelta(hours=val).seconds
+                timeout = int(timedelta(hours=val).total_seconds())
             elif match.group(2) == "d":
                 val = float(match.group(1))
-                timeout = timedelta(days=val).seconds
+                timeout = int(timedelta(days=val).total_seconds())
 
         if not timeout:
             raise AirflowException(

airflow/providers/google/cloud/operators/postgres_to_gcs.py

@@ -91,10 +91,10 @@ def convert_type(self, value, schema_type):
             return time.mktime(value.timetuple())
         if isinstance(value, datetime.time):
             formated_time = time.strptime(str(value), "%H:%M:%S")
-            return datetime.timedelta(
+            return int(datetime.timedelta(
                 hours=formated_time.tm_hour,
                 minutes=formated_time.tm_min,
-                seconds=formated_time.tm_sec).seconds
+                seconds=formated_time.tm_sec).total_seconds())
         if isinstance(value, dict):
             return json.dumps(value)
         if isinstance(value, Decimal):

tests/jobs/test_base_job.py

@@ -95,6 +95,11 @@ def test_is_alive(self):
         job.latest_heartbeat = timezone.utcnow() - datetime.timedelta(seconds=21)
         self.assertFalse(job.is_alive())
 
+        # test because .seconds was used before instead of total_seconds
+        # internal repr of datetime is (days, seconds)
+        job.latest_heartbeat = timezone.utcnow() - datetime.timedelta(days=1)
+        self.assertFalse(job.is_alive())
+
         job.state = State.SUCCESS
         job.latest_heartbeat = timezone.utcnow() - datetime.timedelta(seconds=10)
         self.assertFalse(job.is_alive(), "Completed jobs even with recent heartbeat should not be alive")

tests/jobs/test_scheduler_job.py

@@ -969,6 +969,11 @@ def test_is_alive(self):
         job.latest_heartbeat = timezone.utcnow() - datetime.timedelta(seconds=31)
         self.assertFalse(job.is_alive())
 
+        # test because .seconds was used before instead of total_seconds
+        # internal repr of datetime is (days, seconds)
+        job.latest_heartbeat = timezone.utcnow() - datetime.timedelta(days=1)
+        self.assertFalse(job.is_alive())
+
         job.state = State.SUCCESS
         job.latest_heartbeat = timezone.utcnow() - datetime.timedelta(seconds=10)
         self.assertFalse(job.is_alive(), "Completed jobs even with recent heartbeat should not be alive")