Merge "Do not let a transient job board failure kill the conductor"

releasenotes/notes/bug-2160070-4b7f2a9c1d3e5f60.yaml

@@ -0,0 +1,7 @@
+---
+fixes:
+  - |
+    The executor conductor no longer terminates when a transient job board
+    failure (for example a Redis Sentinel master failover) is raised while
+    iterating or claiming jobs. It now logs the failure, backs off, and
+    continues, reconnecting to the backend on the next cycle.

taskflow/conductors/backends/impl_executor.py

@@ -337,35 +337,47 @@ class ExecutorConductor(base.Conductor, metaclass=abc.ABCMeta):
                     fresh_period.restart()
                 else:
                     ensure_fresh = False
-                job_it = itertools.takewhile(
-                    self._can_claim_more_jobs,
-                    self._jobboard.iterjobs(ensure_fresh=ensure_fresh),
-                )
-                for job in job_it:
-                    self._log.debug("Trying to claim job: %s", job)
-                    try:
-                        self._jobboard.claim(job, self._name)
-                    except (excp.UnclaimableJob, excp.NotFound):
-                        self._log.debug(
-                            "Job already claimed or consumed: %s", job
-                        )
-                    else:
+                try:
+                    job_it = itertools.takewhile(
+                        self._can_claim_more_jobs,
+                        self._jobboard.iterjobs(ensure_fresh=ensure_fresh),
+                    )
+                    for job in job_it:
+                        self._log.debug("Trying to claim job: %s", job)
                         try:
-                            fut = executor.submit(self._dispatch_job, job)
-                        except RuntimeError:
-                            with excutils.save_and_reraise_exception():
-                                self._log.warn(
-                                    "Job dispatch submitting failed: %s", job
-                                )
-                                self._try_finish_job(job, False)
-                        else:
-                            fut.job = job
-                            self._dispatched.add(fut)
-                            any_dispatched = True
-                            fut.add_done_callback(
-                                functools.partial(self._on_job_done, job)
+                            self._jobboard.claim(job, self._name)
+                        except (excp.UnclaimableJob, excp.NotFound):
+                            self._log.debug(
+                                "Job already claimed or consumed: %s", job
                             )
-                            total_dispatched = next(dispatch_gen)
+                        else:
+                            try:
+                                fut = executor.submit(self._dispatch_job, job)
+                            except RuntimeError:
+                                with excutils.save_and_reraise_exception():
+                                    self._log.warn(
+                                        "Job dispatch submitting failed: %s",
+                                        job,
+                                    )
+                                    self._try_finish_job(job, False)
+                            else:
+                                fut.job = job
+                                self._dispatched.add(fut)
+                                any_dispatched = True
+                                fut.add_done_callback(
+                                    functools.partial(self._on_job_done, job)
+                                )
+                                total_dispatched = next(dispatch_gen)
+                except excp.JobFailure:
+                    # A transient job board failure (for example a Redis
+                    # Sentinel master failover) can surface while iterating
+                    # or claiming jobs. Log and back off rather than letting
+                    # it terminate the conductor, which is meant to run until
+                    # stopped; the next cycle reconnects and resumes.
+                    self._log.exception(
+                        "Transient job board failure while iterating jobs; "
+                        "backing off and retrying",
+                    )
                 if not any_dispatched and not is_stopped():
                     self._wait_timeout.wait()
         except StopIteration:

taskflow/tests/unit/test_conductors.py

@@ -23,6 +23,7 @@ import testtools
 
 from taskflow.conductors import backends
 from taskflow import engines
+from taskflow import exceptions as excp
 from taskflow.jobs.backends import impl_zookeeper
 from taskflow.jobs import base
 from taskflow.patterns import linear_flow as lf
@@ -183,6 +184,52 @@ class ManyConductorTest(
         self.assertIsNotNone(fd)
         self.assertEqual(st.SUCCESS, fd.state)
 
+    def test_run_survives_transient_jobboard_failure(self):
+        # Regression test for bug 2160070: a transient JobFailure raised while
+        # iterating the job board must not terminate the conductor.
+        components = self.make_components()
+        components.conductor.connect()
+        consumed_event = threading.Event()
+
+        def on_consume(state, details):
+            consumed_event.set()
+
+        components.board.notifier.register(base.REMOVAL, on_consume)
+
+        # iterjobs() itself does no I/O; it builds a JobBoardIterator and
+        # returns. The backend is only touched later, from inside
+        # JobBoardIterator.__next__, so raise from _fetch_jobs to put the
+        # failure where a real one lands: in the conductor's 'for job in
+        # job_it' statement.
+        real_fetch_jobs = components.board._fetch_jobs
+        state = {'failed_once': False}
+
+        def flaky_fetch_jobs(*args, **kwargs):
+            if not state['failed_once']:
+                state['failed_once'] = True
+                raise excp.JobFailure("simulated transient backend failure")
+            return real_fetch_jobs(*args, **kwargs)
+
+        components.board._fetch_jobs = flaky_fetch_jobs
+
+        with contextlib.closing(components.conductor):
+            t = threading_utils.daemon_thread(components.conductor.run)
+            t.start()
+            lb, fd = pu.temporary_flow_detail(components.persistence)
+            engines.save_factory_details(
+                fd, test_factory, [False], {}, backend=components.persistence
+            )
+            components.board.post('poke', lb, details={'flow_uuid': fd.uuid})
+            # The first job fetch raised JobFailure; with the fix the
+            # conductor backs off and keeps running, so the job is still
+            # consumed. Without the fix, run() dies and this times out.
+            self.assertTrue(consumed_event.wait(test_utils.WAIT_TIMEOUT))
+            self.assertTrue(state['failed_once'])
+            components.conductor.stop()
+            self.assertTrue(components.conductor.wait(test_utils.WAIT_TIMEOUT))
+            self.assertFalse(components.conductor.dispatching)
+            t.join()
+
     def test_run_max_dispatches(self):
         components = self.make_components()
         components.conductor.connect()