self.clear_next_method_args()
            self.end_date = timezone.utcnow()
            _log_state(task_instance=self)
            self.set_duration()
            
            
            
            _run_finished_callback(callbacks=self.task.on_success_callback, context=context)
            if not test_mode:
                session.add(Log(self.state, self))
                session.merge(self).task = self.task
                if self.state == TaskInstanceState.SUCCESS:
                    self._register_dataset_changes(session=session)
                session.commit()
                if self.state == TaskInstanceState.SUCCESS:
                    get_listener_manager().hook.on_task_instance_success(
                        previous_state=TaskInstanceState.RUNNING, task_instance=self, session=session
                    )
            return None
    def _register_dataset_changes(self, *, session: Session) -> None:
        for obj in self.task.outlets or []:
            self.log.debug("outlet obj %s", obj)
            
            if isinstance(obj, Dataset):
                dataset_manager.register_dataset_change(
                    task_instance=self,
                    dataset=obj,
                    session=session,
                )
    def _execute_task_with_callbacks(self, context, test_mode: bool = False, *, session: Session):