Apply suggestions from code review

airflow/models/dagrun.py

@@ -736,7 +736,7 @@ def _filter_tis_and_exclude_removed(dag: "DAG", tis: List[TI]) -> Iterable[TI]:
                     yield ti
 
         tis = list(_filter_tis_and_exclude_removed(self.get_dag(), tis))
-        missing_indexes = self._find_missing_task_indexes_or_mark_removed(tis, session=session)
+        missing_indexes = self._revise_mapped_task_indexes(tis, session=session)
         if missing_indexes:
             self.verify_integrity(missing_indexes=missing_indexes, session=session)
 
@@ -1158,7 +1158,7 @@ def _create_task_instances(
             # TODO[HA]: We probably need to savepoint this so we can keep the transaction alive.
             session.rollback()
 
-    def _find_missing_task_indexes_or_mark_removed(
+    def _revise_mapped_task_indexes(
         self,
         tis: Iterable[TI],
         *,
@@ -1184,7 +1184,7 @@ def _find_missing_task_indexes_or_mark_removed(
             task.run_time_mapped_ti_count.cache_clear()  # type: ignore[attr-defined]
             new_length = task.run_time_mapped_ti_count(self.run_id, session=session) or 0
 
-            if ti.map_index >= new_length and new_length > 0:
+            if ti.map_index >= new_length:
                 self.log.debug(
                     "Removing task '%s' as the map_index is longer than the resolved mapping list (%d)",
                     ti,

tests/models/test_dagrun.py

@@ -1228,7 +1228,10 @@ def task_2(arg2):
 
 
 def test_mapped_literal_length_reduction_at_runtime_adds_removed_state(dag_maker, session):
-    """Test that when the length of mapped literal increases at runtime, additional ti is added"""
+    """
+    Test that when the length of mapped literal reduces at runtime, the missing task instances
+    are marked as removed
+    """
     from airflow.models import Variable
 
     Variable.set(key='arg1', value=[1, 2, 3])