Handle e-mail failures in batch notifier

As there's not much we can do about not sending e-mails, we just ignore
the error. In order to prevent constant e-mail processing, we also
clear all the e-mails from the queue and reset the time period for the
next send.

luigi/batch_notifier.py

@@ -187,19 +187,21 @@ def _send_email(self, fail_counts, disable_counts, scheduling_counts, fail_expls
             send_email(subject, email_body, email().sender, (owner,))
 
     def send_email(self):
-        for owner, failures in six.iteritems(self._fail_counts):
-            self._send_email(
-                fail_counts=failures,
-                disable_counts=self._disabled_counts[owner],
-                scheduling_counts=self._scheduling_fail_counts[owner],
-                fail_expls=self._fail_expls[owner],
-                owner=owner,
-            )
-        self._update_next_send()
-        self._fail_counts.clear()
-        self._disabled_counts.clear()
-        self._scheduling_fail_counts.clear()
-        self._fail_expls.clear()
+        try:
+            for owner, failures in six.iteritems(self._fail_counts):
+                self._send_email(
+                    fail_counts=failures,
+                    disable_counts=self._disabled_counts[owner],
+                    scheduling_counts=self._scheduling_fail_counts[owner],
+                    fail_expls=self._fail_expls[owner],
+                    owner=owner,
+                )
+        finally:
+            self._update_next_send()
+            self._fail_counts.clear()
+            self._disabled_counts.clear()
+            self._scheduling_fail_counts.clear()
+            self._fail_expls.clear()
 
     def update(self):
         if time.time() >= self._next_send:

test/batch_notifier_test.py

@@ -1,4 +1,5 @@
 # coding=utf-8
+from smtplib import SMTPServerDisconnected
 
 import mock
 import unittest
@@ -292,6 +293,17 @@ def test_send_clears_backlog(self):
         bn.send_email()
         self.send_email.assert_not_called()
 
+    def test_email_gets_cleared_on_failure(self):
+        bn = BatchNotifier(batch_mode='all')
+
+        bn.add_failure('Task(a=5)', 'Task', {'a': 1}, '', [])
+        self.send_email.side_effect = SMTPServerDisconnected('timeout')
+        self.assertRaises(SMTPServerDisconnected, bn.send_email)
+
+        self.send_email.reset_mock()
+        bn.send_email()
+        self.send_email.assert_not_called()
+
     def test_send_clears_all_old_data(self):
         bn = BatchNotifier(batch_mode='all', error_messages=100)
 
@@ -362,6 +374,25 @@ def test_no_auto_send_until_end_of_interval_with_error(self):
             '- Task(a=5) (1 failure)'
         )
 
+    def test_no_auto_send_for_interval_after_exception(self):
+        bn = BatchNotifier(batch_mode='all')
+        bn.add_failure('Task(a=5)', 'Task', {'a': 5}, 'error', [])
+        self.send_email.side_effect = SMTPServerDisconnected
+
+        self.incr_time(minutes=60)
+        self.assertRaises(SMTPServerDisconnected, bn.update)
+
+        self.send_email.reset_mock()
+        self.send_email.side_effect = None
+        bn.add_failure('Task(a=5)', 'Task', {'a': 5}, 'error', [])
+        for i in range(60):
+            bn.update()
+            self.send_email.assert_not_called()
+            self.incr_time(minutes=1)
+
+        bn.update()
+        self.assertEqual(1, self.send_email.call_count)
+
     def test_send_batch_failure_emails_to_owners(self):
         bn = BatchNotifier(batch_mode='all')
         bn.add_failure('Task(a=1)', 'Task', {'a': '1'}, 'error', ['a@test.com', 'b@test.com'])