[GitHub] ashb 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>]

ashb 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_r210500785
 
 

 ##########
 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:
   Does SQLAlchemy give us an exception class that we can catch, so that we don't have to deal with all possible engines available to SQLA?

----------------------------------------------------------------
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