openstack/taskflow
Code Issues Proposed changes

Delete the job from backend if it cannot be consumed

Browse Source 
In case of invalid/missing data in the job, the job entry should be
purged from the backend to avoid re-scheduling it later.

Closes-Bug: #1949950

Change-Id: I09ab84883cd4dfbab18b56c61e585fd8ac1b6acf
This commit is contained in:
Gregory Thiemonge 2021-11-05 09:29:48 +01:00 committed by Stephen Finucane
parent 83dfa6581e
commit a45f8249c9
1 changed files with 11 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
taskflow/conductors/backends/impl_executor.py
View File

@ -211,7 +211,7 @@ class ExecutorConductor(base.Conductor):
self._log.info("Job completed successfully: %s", job)
return consume
def _try_finish_job(self, job, consume):
def _try_finish_job(self, job, consume, trash=False):
try:
if consume:
self._jobboard.consume(job, self._name)
@ -220,6 +220,13 @@ class ExecutorConductor(base.Conductor):
'conductor': self,
'persistence': self._persistence,
})
elif trash:
self._jobboard.trash(job, self._name)
self._notifier.notify("job_trashed", {
'job': job,
'conductor': self,
'persistence': self._persistence,
})
else:
self._jobboard.abandon(job, self._name)
self._notifier.notify("job_abandoned", {
@ -237,6 +244,7 @@ class ExecutorConductor(base.Conductor):
def _on_job_done(self, job, fut):
consume = False
trash = False
try:
consume = fut.result()
except KeyboardInterrupt:
@ -244,8 +252,9 @@ class ExecutorConductor(base.Conductor):
self._log.warn("Job dispatching interrupted: %s", job)
except Exception:
self._log.warn("Job dispatching failed: %s", job, exc_info=True)
trash = True
try:
self._try_finish_job(job, consume)
self._try_finish_job(job, consume, trash)
finally:
self._dispatched.discard(fut)