"""Mark the task as deferred and sets up the trigger that is needed to resume it."""
        from airflow.models.trigger import Trigger
        
        trigger_row = Trigger.from_object(defer.trigger)
        session.add(trigger_row)
        session.flush()
        
        
        
        self.state = TaskInstanceState.DEFERRED
        self.trigger_id = trigger_row.id
        self.next_method = defer.method_name
        self.next_kwargs = defer.kwargs or {}
        
        self._try_number -= 1
        
        if defer.timeout is not None:
            self.trigger_timeout = timezone.utcnow() + defer.timeout
        else:
            self.trigger_timeout = None
        
        
        execution_timeout = self.task.execution_timeout
        if execution_timeout:
            if self.trigger_timeout:
                self.trigger_timeout = min(self.start_date + execution_timeout, self.trigger_timeout)
            else:
                self.trigger_timeout = self.start_date + execution_timeout
    def _run_execute_callback(self, context: Context, task: Operator) -> None: