uranusjr commented on code in PR #76420:
URL: https://github.com/apache/airflow/pull/76420#discussion_r4466754336
##########
airflow-core/tests/unit/jobs/test_scheduler_job.py:
##########
@@ -6419,6 +6420,33 @@ def
test_verify_integrity_updates_dag_version_for_waiting_tasks(self, dag_maker,
assert downstream_ti.state == State.SCHEDULED
assert downstream_ti.dag_version_id == latest_version_id
+ def
test_verify_integrity_keeps_mapped_tis_while_cleared_producer_reruns(self,
dag_maker, session):
+ with dag_maker("test_cleared_mapped_producer", session=session):
+ size = EmptyOperator(task_id="size")
+ MockOperator.partial(task_id="mapped").expand(arg1=size.output)
+ dr = dag_maker.create_dagrun(state=DagRunState.RUNNING)
+ assert dr.bundle_version is None
+ size_ti = dr.get_task_instance("size", session=session)
+ size_ti.state = TaskInstanceState.SUCCESS
+ push_mapped_length(size_ti, [1, 2], session=session)
+ dr.task_instance_scheduling_decisions(session=session)
+ for ti in dr.get_task_instances(session=session):
+ ti.state = TaskInstanceState.SUCCESS
+ session.flush()
+ clear_task_instances(dr.get_task_instances(session=session), session)
+
+ with dag_maker("test_cleared_mapped_producer", session=session):
+ size = EmptyOperator(task_id="size")
+ MockOperator.partial(task_id="mapped").expand(arg1=size.output)
+ EmptyOperator(task_id="added")
+
+ self.job_runner = SchedulerJobRunner(job=Job())
+ self.job_runner._schedule_dag_run(dr, session)
+
+ tis = dr.get_task_instances(session=session)
+ assert "added" in {ti.task_id for ti in tis}
+ assert {ti.map_index: ti.state for ti in tis if ti.task_id ==
"mapped"} == {0: None, 1: None}
Review Comment:
Consider also push the producer value and assert mapped indexes are
scheduled, similar to
`test_clear_latest_with_producer_keeps_mapped_task_for_revision`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
