Repository: apache/airflow
File: providers/standard/src/airflow/providers/standard/operators/trigger_dagrun.py (Python)

Code before:
        # Save run_id as task attribute - to be used by listeners
        self.trigger_run_id = run_id

        if self.fail_when_dag_is_paused:
            dag_model = DagModel.get_current(self.trigger_dag_id)
            if not dag_model:
                raise ValueError(f"Dag {self.trigger_dag_id} is not found")
            if dag_model.is_paused:
                # TODO: enable this when dag state endpoint available from task sdk
                # if AIRFLOW_V_3_0_PLUS:
                #     raise DagIsPaused(dag_id=self.trigger_dag_id)
                raise AirflowException(f"Dag {self.trigger_dag_id} is paused")

        if AIRFLOW_V_3_0_PLUS:
            self._trigger_dag_af_3(
                context=context, run_id=self.trigger_run_id, parsed_logical_date=parsed_logical_date
            )
        else:
            self._trigger_dag_af_2(
                context=context, run_id=self.trigger_run_id, parsed_logical_date=parsed_logical_date
            )

    def _trigger_dag_af_3(self, context, run_id, parsed_logical_date):
        from airflow.providers.common.compat.sdk import DagRunTriggerException

        kwargs_accepted = dict(
            trigger_dag_id=self.trigger_dag_id,
            dag_run_id=run_id,
            conf=self.conf,
            logical_date=parsed_logical_date,
            reset_dag_run=self.reset_dag_run,
            skip_when_already_exists=self.skip_when_already_exists,
            wait_for_completion=self.wait_for_completion,
            allowed_states=self.allowed_states,
            failed_states=self.failed_states,
            poke_interval=self.poke_interval,
            deferrable=self.deferrable,
        )

        if self.note:
            sig = inspect.signature(DagRunTriggerException.__init__)
            if "note" in sig.parameters:
                kwargs_accepted["note"] = self.note

        raise DagRunTriggerException(**kwargs_accepted)

    def _trigger_dag_af_2(self, context, run_id, parsed_logical_date):
        try:
            dag_run = trigger_dag(
                dag_id=self.trigger_dag_id,
                run_id=run_id,

Review comment (suggestion):
One question: For `_trigger_dag_af_2` path, do we need to handle `note`?
e.g. If `note` is set for AF2, we could log warning for user.

Code after:
        # Save run_id as task attribute - to be used by listeners
        self.trigger_run_id = run_id

        if self.fail_when_dag_is_paused:
            dag_model = DagModel.get_current(self.trigger_dag_id)
            if not dag_model:
                raise ValueError(f"Dag {self.trigger_dag_id} is not found")
            if dag_model.is_paused:
                # TODO: enable this when dag state endpoint available from task sdk
                # if AIRFLOW_V_3_0_PLUS:
                #     raise DagIsPaused(dag_id=self.trigger_dag_id)
                raise AirflowException(f"Dag {self.trigger_dag_id} is paused")

        if AIRFLOW_V_3_0_PLUS:
            self._trigger_dag_af_3(
                context=context, run_id=self.trigger_run_id, parsed_logical_date=parsed_logical_date
            )
        else:
            self._trigger_dag_af_2(
                context=context, run_id=self.trigger_run_id, parsed_logical_date=parsed_logical_date
            )

    def _trigger_dag_af_3(self, context, run_id, parsed_logical_date):
        from airflow.providers.common.compat.sdk import DagRunTriggerException

        kwargs_accepted = dict(
            trigger_dag_id=self.trigger_dag_id,
            dag_run_id=run_id,
            conf=self.conf,
            logical_date=parsed_logical_date,
            reset_dag_run=self.reset_dag_run,
            skip_when_already_exists=self.skip_when_already_exists,
            wait_for_completion=self.wait_for_completion,
            allowed_states=self.allowed_states,
            failed_states=self.failed_states,
            poke_interval=self.poke_interval,
            deferrable=self.deferrable,
        )

        if self.note and "note" in inspect.signature(DagRunTriggerException.__init__).parameters:
            kwargs_accepted["note"] = self.note

        raise DagRunTriggerException(**kwargs_accepted)

    def _trigger_dag_af_2(self, context, run_id, parsed_logical_date):
        try:
            if self.note:
                self.log.warning("Parameter 'note' is not supported in Airflow 2.x and will be ignored.")

            dag_run = trigger_dag(
                dag_id=self.trigger_dag_id,