openstack/taskflow
Code Issues Proposed changes

Merge "Delete the job from backend if it cannot be consumed"

Browse Source
This commit is contained in:
Zuul 2024-01-29 12:46:02 +00:00 committed by Gerrit Code Review
commit b46a6c8b6d
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

@ -209,7 +209,7 @@ class ExecutorConductor(base.Conductor, metaclass=abc.ABCMeta):
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)
@ -218,6 +218,13 @@ class ExecutorConductor(base.Conductor, metaclass=abc.ABCMeta):
'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", {
@ -235,6 +242,7 @@ class ExecutorConductor(base.Conductor, metaclass=abc.ABCMeta):
def _on_job_done(self, job, fut):
consume = False
trash = False
try:
consume = fut.result()
except KeyboardInterrupt:
@ -242,8 +250,9 @@ class ExecutorConductor(base.Conductor, metaclass=abc.ABCMeta):
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)