diff --git a/airflow/models/dagrun.py b/airflow/models/dagrun.py
index 64c27303a1f6a..fde2aa7685ed5 100644
--- a/airflow/models/dagrun.py
+++ b/airflow/models/dagrun.py
@@ -728,9 +728,10 @@ def _filter_tis_and_exclude_removed(dag: "DAG", tis: List[TI]) -> Iterable[TI]:
                 try:
                     ti.task = dag.get_task(ti.task_id)
                 except TaskNotFound:
-                    self.log.error("Failed to get task for ti %s. Marking it as removed.", ti)
-                    ti.state = State.REMOVED
-                    session.flush()
+                    if ti.state != State.REMOVED:
+                        self.log.error("Failed to get task for ti %s. Marking it as removed.", ti)
+                        ti.state = State.REMOVED
+                        session.flush()
                 else:
                     yield ti