[GitHub] feng-tao commented on a change in pull request #3650: [AIRFLOW-2703] exceptions from scheduler's heartbeat is handled so that scheduler does not crash

[GitBox <gi...@apache.org>]

feng-tao commented on a change in pull request #3650: [AIRFLOW-2703] exceptions from scheduler's heartbeat is handled so that scheduler does not crash
URL: https://github.com/apache/incubator-airflow/pull/3650#discussion_r207726248
 
 

 ##########
 File path: airflow/jobs.py
 ##########
 @@ -160,33 +160,36 @@ def heartbeat(self):
         heart rate. If you go over 60 seconds before calling it, it won't
         sleep at all.
         """
-        with create_session() as session:
-            job = session.query(BaseJob).filter_by(id=self.id).one()
-            make_transient(job)
-            session.commit()
+        try:
+            with create_session() as session:
+                job = session.query(BaseJob).filter_by(id=self.id).one()
+                make_transient(job)
+                session.commit()
 
-        if job.state == State.SHUTDOWN:
-            self.kill()
+            if job.state == State.SHUTDOWN:
+                self.kill()
 
-        # Figure out how long to sleep for
-        sleep_for = 0
-        if job.latest_heartbeat:
-            sleep_for = max(
-                0,
-                self.heartrate - (
-                    timezone.utcnow() - job.latest_heartbeat).total_seconds())
+            # Figure out how long to sleep for
+            sleep_for = 0
+            if job.latest_heartbeat:
+                sleep_for = max(
+                    0,
+                    self.heartrate - (timezone.utcnow() -
+                                      job.latest_heartbeat).total_seconds())
 
-        sleep(sleep_for)
+            sleep(sleep_for)
 
-        # Update last heartbeat time
-        with create_session() as session:
-            job = session.query(BaseJob).filter(BaseJob.id == self.id).first()
-            job.latest_heartbeat = timezone.utcnow()
-            session.merge(job)
-            session.commit()
+            # Update last heartbeat time
+            with create_session() as session:
+                job = session.query(BaseJob).filter(BaseJob.id == self.id).first()
+                job.latest_heartbeat = timezone.utcnow()
+                session.merge(job)
+                session.commit()
 
-            self.heartbeat_callback(session=session)
-            self.log.debug('[heartbeat]')
+                self.heartbeat_callback(session=session)
+                self.log.debug('[heartbeat]')
+        except Exception as e:
 
 Review comment:
   exception here is too board. What exception are we expecting? I think we need to be a little bit more specific.
   And scheduler could always get restarted by runit. 

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services