docs: def _complete_task(self, task_name, **kwargs):
          """ Marks this task as completed. Kwargs are stored in the run log. """
  
          logger.debug('Job {0} marking task {1} as completed'.format(self.name, task_name))
          self.run_log['tasks'][task_name] = kwargs
  
          for node in self.downstream(task_name, self.snapshot):
              self._start_if_ready(node)
  
          try:
              self.backend.acquire_lock()
              self._commit_run_log()
          except:
              logger.exception("Error in handling events.")
          finally:
              self.backend.release_lock()
  
          if kwargs.get('success', None) == False:
              task = self.tasks[task_name]
              try:
                  self.backend.acquire_lock()
                  if self.event_handler:
                      self.event_handler.emit('task_failed',
                                              task._serialize(include_run_logs=True))
              except:
                  logger.exception("Error in handling events.")
              finally:
                  self.backend.release_lock()
  
          self._on_completion()