Add start date to trigger_dagrun operator

[bhavaniravi]

closes: #18082

---

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code change, Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in UPDATING.md.

[ephraimbuddy]

I think that you made the change in the wrong place. Here's where I think the change should be:

airflow/airflow/operators/trigger_dagrun.py

Line 132 in 23a68fa

conf=self.conf,

Add start_date=self.start_date after the line above.

EDIT:
You are actually correct, just fix the failing tests

[github-actions]

The PR most likely needs to run full matrix of tests because it modifies parts of the core of Airflow. However, committers might decide to merge it quickly and take the risk. If they don't merge it quickly - please rebase it to the latest main at your convenience, or amend the last commit of the PR, and push it with --force-with-lease.

[kaxil]

=========================== short test summary info ============================
  FAILED tests/models/test_cleartasks.py::TestClearTasks::test_clear_task_instances
  FAILED tests/models/test_cleartasks.py::TestClearTasks::test_clear_task_instances_with_task_reschedule
  FAILED tests/models/test_cleartasks.py::TestClearTasks::test_dag_clear - asse...
  = 3 failed, 2445 passed, 61 skipped, 3 xfailed, 4 warnings in 508.01s (0:08:28) =

tests are failing

[bhavaniravi]

Hey, @kaxil, I cannot recreate the failing tests using breeze env or pytest with the same python and Postgres versions. Looks like a persistent issue across PRs,

https://github.com/apache/airflow/pull/18243/checks?check_run_id=3604451375
Let me rebase and push again

[ephraimbuddy]

Looks like something went wrong in your last rebase. Please rebase again

[alexInhert]

If the dag you trigger is timezone aware does it matter?
I mean you consider here the default timezone but the dag you are triggering may be in a diffrent timezone than the default

[bhavaniravi]

The start_date here is not the DAG start_date. It is the dag run starting time of the run. Internally it's all tracked in UTC and rendered in the user's timezone on UI.

Correct me if I am wrong.

[uranusjr]

You’re correct. Although maybe we shouldn’t set this to RUNNING in the first place, but should set it to QUEUED and let the scheduler start the run (and set start_date) instead. See how this is implemented in the new REST API (in airflow/api_connexion/endpoints/dag_run_endpoint.py::post_dag_run).

[ephraimbuddy]

Supported. Setting it to queued will save us some complaints especially about max_active_runs

[bhavaniravi]

Should I go ahead and make that change? There are 3 places where we are creating dag_runs with the state running, should this be the same behavior across the board?

[ephraimbuddy]

We should only change the one identified here. Looks like others are more involved change and requires a separate PR(that's if it should be changed)

[bhavaniravi]

This change will affect three places. All these use the trigger_dag function.

1. TriggerDAGRunOperator
2. APIClient - dag_runs
3. CLI trigger dag

[kaxil]

Unit tests are failing - https://github.com/apache/airflow/pull/18226/checks?check_run_id=3651105087#step:6:6287

[uranusjr]

Failure seems unrelated

   _____________________ test_external_task_marker_exception ______________________
  
  dag_bag_ext = <airflow.models.dagbag.DagBag object at 0x7fec5c30c730>
  
      def test_external_task_marker_exception(dag_bag_ext):
          """
          Clearing across multiple DAGs should raise AirflowException if more levels are being cleared
          than allowed by the recursion_depth of the first ExternalTaskMarker being cleared.
          """
  >       run_tasks(dag_bag_ext)
  ...
  >       assert task_instance.state == state
  E       AssertionError: assert None == <TaskInstanceState.SUCCESS: 'success'>
  E        +  where None = <TaskInstance: dag_0.task_b_0 manual__2015-01-01T00:00:00+00:00 [None]>.state

[uranusjr]

You probably want to add an explicit order_by here, the db does not always return rows in a deterministic order.

[kaxil]

Thanks @bhavaniravi 🎉